hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTable.java

   1 /**
   2  * Licensed to the Apache Software Foundation (ASF) under one
   3  * or more contributor license agreements.  See the NOTICE file
   4  * distributed with this work for additional information
   5  * regarding copyright ownership.  The ASF licenses this file
   6  * to you under the Apache License, Version 2.0 (the
   7  * "License"); you may not use this file except in compliance
   8  * with the License.  You may obtain a copy of the License at
   9  *
  10  *     http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  */
  18 package org.apache.hadoop.hbase.client;
  19
  20 import static org.hamcrest.CoreMatchers.containsString;
  21 import static org.hamcrest.CoreMatchers.instanceOf;
  22 import static org.hamcrest.MatcherAssert.assertThat;
  23 import static org.junit.Assert.assertArrayEquals;
  24 import static org.junit.Assert.assertEquals;
  25 import static org.junit.Assert.assertFalse;
  26 import static org.junit.Assert.assertNull;
  27 import static org.junit.Assert.assertTrue;
  28 import static org.junit.Assert.fail;
  29
  30 import java.io.IOException;
  31 import java.io.UncheckedIOException;
  32 import java.util.Arrays;
  33 import java.util.Collections;
  34 import java.util.List;
  35 import java.util.concurrent.ArrayBlockingQueue;
  36 import java.util.concurrent.BlockingQueue;
  37 import java.util.concurrent.CountDownLatch;
  38 import java.util.concurrent.ExecutionException;
  39 import java.util.concurrent.ForkJoinPool;
  40 import java.util.concurrent.atomic.AtomicInteger;
  41 import java.util.concurrent.atomic.AtomicLong;
  42 import java.util.function.Supplier;
  43 import java.util.stream.IntStream;
  44 import org.apache.hadoop.hbase.CompareOperator;
  45 import org.apache.hadoop.hbase.HBaseClassTestRule;
  46 import org.apache.hadoop.hbase.HBaseTestingUtil;
  47 import org.apache.hadoop.hbase.TableName;
  48 import org.apache.hadoop.hbase.TableNotEnabledException;
  49 import org.apache.hadoop.hbase.filter.BinaryComparator;
  50 import org.apache.hadoop.hbase.filter.FamilyFilter;
  51 import org.apache.hadoop.hbase.filter.FilterList;
  52 import org.apache.hadoop.hbase.filter.QualifierFilter;
  53 import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
  54 import org.apache.hadoop.hbase.filter.TimestampsFilter;
  55 import org.apache.hadoop.hbase.io.TimeRange;
  56 import org.apache.hadoop.hbase.testclassification.ClientTests;
  57 import org.apache.hadoop.hbase.testclassification.MediumTests;
  58 import org.apache.hadoop.hbase.util.Bytes;
  59 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
  60 import org.apache.hadoop.hbase.util.Pair;
  61 import org.junit.AfterClass;
  62 import org.junit.Before;
  63 import org.junit.BeforeClass;
  64 import org.junit.ClassRule;
  65 import org.junit.Rule;
  66 import org.junit.Test;
  67 import org.junit.experimental.categories.Category;
  68 import org.junit.rules.TestName;
  69 import org.junit.runner.RunWith;
  70 import org.junit.runners.Parameterized;
  71 import org.junit.runners.Parameterized.Parameter;
  72 import org.junit.runners.Parameterized.Parameters;
  73
  74 import org.apache.hbase.thirdparty.com.google.common.io.Closeables;
  75
  76 @RunWith(Parameterized.class)
  77 @Category({ MediumTests.class, ClientTests.class })
  78 public class TestAsyncTable {
  79
  80   @ClassRule
  81   public static final HBaseClassTestRule CLASS_RULE =
  82     HBaseClassTestRule.forClass(TestAsyncTable.class);
  83
  84   private static final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
  85
  86   private static TableName TABLE_NAME = TableName.valueOf("async");
  87
  88   private static byte[] FAMILY = Bytes.toBytes("cf");
  89
  90   private static byte[] QUALIFIER = Bytes.toBytes("cq");
  91
  92   private static byte[] VALUE = Bytes.toBytes("value");
  93
  94   private static int MAX_KEY_VALUE_SIZE = 64 * 1024;
  95
  96   private static AsyncConnection ASYNC_CONN;
  97
  98   @Rule
  99   public TestName testName = new TestName();
 100
 101   private byte[] row;
 102
 103   @Parameter
 104   public Supplier<AsyncTable<?>> getTable;
 105
 106   private static AsyncTable<?> getRawTable() {
 107     return ASYNC_CONN.getTable(TABLE_NAME);
 108   }
 109
 110   private static AsyncTable<?> getTable() {
 111     return ASYNC_CONN.getTable(TABLE_NAME, ForkJoinPool.commonPool());
 112   }
 113
 114   @Parameters
 115   public static List<Object[]> params() {
 116     return Arrays.asList(new Supplier<?>[] { TestAsyncTable::getRawTable },
 117       new Supplier<?>[] { TestAsyncTable::getTable });
 118   }
 119
 120   @BeforeClass
 121   public static void setUpBeforeClass() throws Exception {
 122     TEST_UTIL.getConfiguration().setInt(ConnectionConfiguration.MAX_KEYVALUE_SIZE_KEY,
 123       MAX_KEY_VALUE_SIZE);
 124     TEST_UTIL.startMiniCluster(1);
 125     TEST_UTIL.createTable(TABLE_NAME, FAMILY);
 126     TEST_UTIL.waitTableAvailable(TABLE_NAME);
 127     ASYNC_CONN = ConnectionFactory.createAsyncConnection(TEST_UTIL.getConfiguration()).get();
 128     assertFalse(ASYNC_CONN.isClosed());
 129   }
 130
 131   @AfterClass
 132   public static void tearDownAfterClass() throws Exception {
 133     Closeables.close(ASYNC_CONN, true);
 134     assertTrue(ASYNC_CONN.isClosed());
 135     TEST_UTIL.shutdownMiniCluster();
 136   }
 137
 138   @Before
 139   public void setUp() throws IOException, InterruptedException, ExecutionException {
 140     row = Bytes.toBytes(testName.getMethodName().replaceAll("[^0-9A-Za-z]", "_"));
 141     if (ASYNC_CONN.getAdmin().isTableDisabled(TABLE_NAME).get()) {
 142       ASYNC_CONN.getAdmin().enableTable(TABLE_NAME).get();
 143     }
 144   }
 145
 146   @Test
 147   public void testSimple() throws Exception {
 148     AsyncTable<?> table = getTable.get();
 149     table.put(new Put(row).addColumn(FAMILY, QUALIFIER, VALUE)).get();
 150     assertTrue(table.exists(new Get(row).addColumn(FAMILY, QUALIFIER)).get());
 151     Result result = table.get(new Get(row).addColumn(FAMILY, QUALIFIER)).get();
 152     assertArrayEquals(VALUE, result.getValue(FAMILY, QUALIFIER));
 153     table.delete(new Delete(row)).get();
 154     result = table.get(new Get(row).addColumn(FAMILY, QUALIFIER)).get();
 155     assertTrue(result.isEmpty());
 156     assertFalse(table.exists(new Get(row).addColumn(FAMILY, QUALIFIER)).get());
 157   }
 158
 159   private byte[] concat(byte[] base, int index) {
 160     return Bytes.toBytes(Bytes.toString(base) + "-" + index);
 161   }
 162
 163   @SuppressWarnings("FutureReturnValueIgnored")
 164   @Test
 165   public void testSimpleMultiple() throws Exception {
 166     AsyncTable<?> table = getTable.get();
 167     int count = 100;
 168     CountDownLatch putLatch = new CountDownLatch(count);
 169     IntStream.range(0, count).forEach(
 170       i -> table.put(new Put(concat(row, i)).addColumn(FAMILY, QUALIFIER, concat(VALUE, i)))
 171         .thenAccept(x -> putLatch.countDown()));
 172     putLatch.await();
 173     BlockingQueue<Boolean> existsResp = new ArrayBlockingQueue<>(count);
 174     IntStream.range(0, count)
 175       .forEach(i -> table.exists(new Get(concat(row, i)).addColumn(FAMILY, QUALIFIER))
 176         .thenAccept(x -> existsResp.add(x)));
 177     for (int i = 0; i < count; i++) {
 178       assertTrue(existsResp.take());
 179     }
 180     BlockingQueue<Pair<Integer, Result>> getResp = new ArrayBlockingQueue<>(count);
 181     IntStream.range(0, count)
 182       .forEach(i -> table.get(new Get(concat(row, i)).addColumn(FAMILY, QUALIFIER))
 183         .thenAccept(x -> getResp.add(Pair.newPair(i, x))));
 184     for (int i = 0; i < count; i++) {
 185       Pair<Integer, Result> pair = getResp.take();
 186       assertArrayEquals(concat(VALUE, pair.getFirst()),
 187         pair.getSecond().getValue(FAMILY, QUALIFIER));
 188     }
 189     CountDownLatch deleteLatch = new CountDownLatch(count);
 190     IntStream.range(0, count).forEach(
 191       i -> table.delete(new Delete(concat(row, i))).thenAccept(x -> deleteLatch.countDown()));
 192     deleteLatch.await();
 193     IntStream.range(0, count)
 194       .forEach(i -> table.exists(new Get(concat(row, i)).addColumn(FAMILY, QUALIFIER))
 195         .thenAccept(x -> existsResp.add(x)));
 196     for (int i = 0; i < count; i++) {
 197       assertFalse(existsResp.take());
 198     }
 199     IntStream.range(0, count)
 200       .forEach(i -> table.get(new Get(concat(row, i)).addColumn(FAMILY, QUALIFIER))
 201         .thenAccept(x -> getResp.add(Pair.newPair(i, x))));
 202     for (int i = 0; i < count; i++) {
 203       Pair<Integer, Result> pair = getResp.take();
 204       assertTrue(pair.getSecond().isEmpty());
 205     }
 206   }
 207
 208   @SuppressWarnings("FutureReturnValueIgnored")
 209   @Test
 210   public void testIncrement() throws InterruptedException, ExecutionException {
 211     AsyncTable<?> table = getTable.get();
 212     int count = 100;
 213     CountDownLatch latch = new CountDownLatch(count);
 214     AtomicLong sum = new AtomicLong(0L);
 215     IntStream.range(0, count)
 216       .forEach(i -> table.incrementColumnValue(row, FAMILY, QUALIFIER, 1).thenAccept(x -> {
 217         sum.addAndGet(x);
 218         latch.countDown();
 219       }));
 220     latch.await();
 221     assertEquals(count, Bytes.toLong(
 222       table.get(new Get(row).addColumn(FAMILY, QUALIFIER)).get().getValue(FAMILY, QUALIFIER)));
 223     assertEquals((1 + count) * count / 2, sum.get());
 224   }
 225
 226   @SuppressWarnings("FutureReturnValueIgnored")
 227   @Test
 228   public void testAppend() throws InterruptedException, ExecutionException {
 229     AsyncTable<?> table = getTable.get();
 230     int count = 10;
 231     CountDownLatch latch = new CountDownLatch(count);
 232     char suffix = ':';
 233     AtomicLong suffixCount = new AtomicLong(0L);
 234     IntStream.range(0, count)
 235       .forEachOrdered(i -> table
 236         .append(new Append(row).addColumn(FAMILY, QUALIFIER, Bytes.toBytes("" + i + suffix)))
 237         .thenAccept(r -> {
 238           suffixCount.addAndGet(
 239             Bytes.toString(r.getValue(FAMILY, QUALIFIER)).chars().filter(x -> x == suffix).count());
 240           latch.countDown();
 241         }));
 242     latch.await();
 243     assertEquals((1 + count) * count / 2, suffixCount.get());
 244     String value = Bytes.toString(
 245       table.get(new Get(row).addColumn(FAMILY, QUALIFIER)).get().getValue(FAMILY, QUALIFIER));
 246     int[] actual = Arrays.asList(value.split("" + suffix)).stream().mapToInt(Integer::parseInt)
 247       .sorted().toArray();
 248     assertArrayEquals(IntStream.range(0, count).toArray(), actual);
 249   }
 250
 251   @Test
 252   public void testMutateRow() throws InterruptedException, ExecutionException, IOException {
 253     AsyncTable<?> table = getTable.get();
 254     RowMutations mutation = new RowMutations(row);
 255     mutation.add(new Put(row).addColumn(FAMILY, concat(QUALIFIER, 1), VALUE));
 256     Result result = table.mutateRow(mutation).get();
 257     assertTrue(result.getExists());
 258     assertTrue(result.isEmpty());
 259
 260     result = table.get(new Get(row)).get();
 261     assertArrayEquals(VALUE, result.getValue(FAMILY, concat(QUALIFIER, 1)));
 262
 263     mutation = new RowMutations(row);
 264     mutation.add(new Delete(row).addColumn(FAMILY, concat(QUALIFIER, 1)));
 265     mutation.add(new Put(row).addColumn(FAMILY, concat(QUALIFIER, 2), VALUE));
 266     mutation.add(new Increment(row).addColumn(FAMILY, concat(QUALIFIER, 3), 2L));
 267     mutation.add(new Append(row).addColumn(FAMILY, concat(QUALIFIER, 4), Bytes.toBytes("abc")));
 268     result = table.mutateRow(mutation).get();
 269     assertTrue(result.getExists());
 270     assertEquals(2L, Bytes.toLong(result.getValue(FAMILY, concat(QUALIFIER, 3))));
 271     assertEquals("abc", Bytes.toString(result.getValue(FAMILY, concat(QUALIFIER, 4))));
 272
 273     result = table.get(new Get(row)).get();
 274     assertNull(result.getValue(FAMILY, concat(QUALIFIER, 1)));
 275     assertArrayEquals(VALUE, result.getValue(FAMILY, concat(QUALIFIER, 2)));
 276     assertEquals(2L, Bytes.toLong(result.getValue(FAMILY, concat(QUALIFIER, 3))));
 277     assertEquals("abc", Bytes.toString(result.getValue(FAMILY, concat(QUALIFIER, 4))));
 278   }
 279
 280   // Tests for old checkAndMutate API
 281
 282   @SuppressWarnings("FutureReturnValueIgnored")
 283   @Test
 284   @Deprecated
 285   public void testCheckAndPutForOldApi() throws InterruptedException, ExecutionException {
 286     AsyncTable<?> table = getTable.get();
 287     AtomicInteger successCount = new AtomicInteger(0);
 288     AtomicInteger successIndex = new AtomicInteger(-1);
 289     int count = 10;
 290     CountDownLatch latch = new CountDownLatch(count);
 291     IntStream.range(0, count)
 292       .forEach(i -> table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER).ifNotExists()
 293         .thenPut(new Put(row).addColumn(FAMILY, QUALIFIER, concat(VALUE, i))).thenAccept(x -> {
 294           if (x) {
 295             successCount.incrementAndGet();
 296             successIndex.set(i);
 297           }
 298           latch.countDown();
 299         }));
 300     latch.await();
 301     assertEquals(1, successCount.get());
 302     String actual = Bytes.toString(table.get(new Get(row)).get().getValue(FAMILY, QUALIFIER));
 303     assertTrue(actual.endsWith(Integer.toString(successIndex.get())));
 304   }
 305
 306   @SuppressWarnings("FutureReturnValueIgnored")
 307   @Test
 308   @Deprecated
 309   public void testCheckAndDeleteForOldApi() throws InterruptedException, ExecutionException {
 310     AsyncTable<?> table = getTable.get();
 311     int count = 10;
 312     CountDownLatch putLatch = new CountDownLatch(count + 1);
 313     table.put(new Put(row).addColumn(FAMILY, QUALIFIER, VALUE)).thenRun(() -> putLatch.countDown());
 314     IntStream.range(0, count)
 315       .forEach(i -> table.put(new Put(row).addColumn(FAMILY, concat(QUALIFIER, i), VALUE))
 316         .thenRun(() -> putLatch.countDown()));
 317     putLatch.await();
 318
 319     AtomicInteger successCount = new AtomicInteger(0);
 320     AtomicInteger successIndex = new AtomicInteger(-1);
 321     CountDownLatch deleteLatch = new CountDownLatch(count);
 322     IntStream.range(0, count)
 323       .forEach(i -> table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER).ifEquals(VALUE)
 324         .thenDelete(
 325           new Delete(row).addColumn(FAMILY, QUALIFIER).addColumn(FAMILY, concat(QUALIFIER, i)))
 326         .thenAccept(x -> {
 327           if (x) {
 328             successCount.incrementAndGet();
 329             successIndex.set(i);
 330           }
 331           deleteLatch.countDown();
 332         }));
 333     deleteLatch.await();
 334     assertEquals(1, successCount.get());
 335     Result result = table.get(new Get(row)).get();
 336     IntStream.range(0, count).forEach(i -> {
 337       if (i == successIndex.get()) {
 338         assertFalse(result.containsColumn(FAMILY, concat(QUALIFIER, i)));
 339       } else {
 340         assertArrayEquals(VALUE, result.getValue(FAMILY, concat(QUALIFIER, i)));
 341       }
 342     });
 343   }
 344
 345   @SuppressWarnings("FutureReturnValueIgnored")
 346   @Test
 347   @Deprecated
 348   public void testCheckAndMutateForOldApi() throws InterruptedException, ExecutionException {
 349     AsyncTable<?> table = getTable.get();
 350     int count = 10;
 351     CountDownLatch putLatch = new CountDownLatch(count + 1);
 352     table.put(new Put(row).addColumn(FAMILY, QUALIFIER, VALUE)).thenRun(() -> putLatch.countDown());
 353     IntStream.range(0, count)
 354       .forEach(i -> table.put(new Put(row).addColumn(FAMILY, concat(QUALIFIER, i), VALUE))
 355         .thenRun(() -> putLatch.countDown()));
 356     putLatch.await();
 357
 358     AtomicInteger successCount = new AtomicInteger(0);
 359     AtomicInteger successIndex = new AtomicInteger(-1);
 360     CountDownLatch mutateLatch = new CountDownLatch(count);
 361     IntStream.range(0, count).forEach(i -> {
 362       RowMutations mutation = new RowMutations(row);
 363       try {
 364         mutation.add((Mutation) new Delete(row).addColumn(FAMILY, QUALIFIER));
 365         mutation
 366           .add((Mutation) new Put(row).addColumn(FAMILY, concat(QUALIFIER, i), concat(VALUE, i)));
 367       } catch (IOException e) {
 368         throw new UncheckedIOException(e);
 369       }
 370       table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER).ifEquals(VALUE).thenMutate(mutation)
 371         .thenAccept(x -> {
 372           if (x) {
 373             successCount.incrementAndGet();
 374             successIndex.set(i);
 375           }
 376           mutateLatch.countDown();
 377         });
 378     });
 379     mutateLatch.await();
 380     assertEquals(1, successCount.get());
 381     Result result = table.get(new Get(row)).get();
 382     IntStream.range(0, count).forEach(i -> {
 383       if (i == successIndex.get()) {
 384         assertArrayEquals(concat(VALUE, i), result.getValue(FAMILY, concat(QUALIFIER, i)));
 385       } else {
 386         assertArrayEquals(VALUE, result.getValue(FAMILY, concat(QUALIFIER, i)));
 387       }
 388     });
 389   }
 390
 391   @Test
 392   @Deprecated
 393   public void testCheckAndMutateWithTimeRangeForOldApi() throws Exception {
 394     AsyncTable<?> table = getTable.get();
 395     final long ts = EnvironmentEdgeManager.currentTime() / 2;
 396     Put put = new Put(row);
 397     put.addColumn(FAMILY, QUALIFIER, ts, VALUE);
 398
 399     boolean ok =
 400       table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER).ifNotExists().thenPut(put).get();
 401     assertTrue(ok);
 402
 403     ok = table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER).timeRange(TimeRange.at(ts + 10000))
 404       .ifEquals(VALUE).thenPut(put).get();
 405     assertFalse(ok);
 406
 407     ok = table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER).timeRange(TimeRange.at(ts))
 408       .ifEquals(VALUE).thenPut(put).get();
 409     assertTrue(ok);
 410
 411     RowMutations rm = new RowMutations(row).add((Mutation) put);
 412
 413     ok = table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER).timeRange(TimeRange.at(ts + 10000))
 414       .ifEquals(VALUE).thenMutate(rm).get();
 415     assertFalse(ok);
 416
 417     ok = table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER).timeRange(TimeRange.at(ts))
 418       .ifEquals(VALUE).thenMutate(rm).get();
 419     assertTrue(ok);
 420
 421     Delete delete = new Delete(row).addColumn(FAMILY, QUALIFIER);
 422
 423     ok = table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER).timeRange(TimeRange.at(ts + 10000))
 424       .ifEquals(VALUE).thenDelete(delete).get();
 425     assertFalse(ok);
 426
 427     ok = table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER).timeRange(TimeRange.at(ts))
 428       .ifEquals(VALUE).thenDelete(delete).get();
 429     assertTrue(ok);
 430   }
 431
 432   @Test
 433   @Deprecated
 434   public void testCheckAndMutateWithSingleFilterForOldApi() throws Throwable {
 435     AsyncTable<?> table = getTable.get();
 436
 437     // Put one row
 438     Put put = new Put(row);
 439     put.addColumn(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a"));
 440     put.addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("b"));
 441     put.addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c"));
 442     table.put(put).get();
 443
 444     // Put with success
 445     boolean ok = table.checkAndMutate(row, new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"),
 446         CompareOperator.EQUAL, Bytes.toBytes("a")))
 447       .thenPut(new Put(row).addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("d")))
 448       .get();
 449     assertTrue(ok);
 450
 451     Result result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("D"))).get();
 452     assertEquals("d", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("D"))));
 453
 454     // Put with failure
 455     ok = table.checkAndMutate(row, new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"),
 456         CompareOperator.EQUAL, Bytes.toBytes("b")))
 457       .thenPut(new Put(row).addColumn(FAMILY, Bytes.toBytes("E"), Bytes.toBytes("e")))
 458       .get();
 459     assertFalse(ok);
 460
 461     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("E"))).get());
 462
 463     // Delete with success
 464     ok = table.checkAndMutate(row, new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"),
 465         CompareOperator.EQUAL, Bytes.toBytes("a")))
 466       .thenDelete(new Delete(row).addColumns(FAMILY, Bytes.toBytes("D")))
 467       .get();
 468     assertTrue(ok);
 469
 470     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("D"))).get());
 471
 472     // Mutate with success
 473     ok = table.checkAndMutate(row, new SingleColumnValueFilter(FAMILY, Bytes.toBytes("B"),
 474         CompareOperator.EQUAL, Bytes.toBytes("b")))
 475       .thenMutate(new RowMutations(row)
 476         .add((Mutation) new Put(row)
 477           .addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("d")))
 478         .add((Mutation) new Delete(row).addColumns(FAMILY, Bytes.toBytes("A"))))
 479       .get();
 480     assertTrue(ok);
 481
 482     result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("D"))).get();
 483     assertEquals("d", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("D"))));
 484
 485     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("A"))).get());
 486   }
 487
 488   @Test
 489   @Deprecated
 490   public void testCheckAndMutateWithMultipleFiltersForOldApi() throws Throwable {
 491     AsyncTable<?> table = getTable.get();
 492
 493     // Put one row
 494     Put put = new Put(row);
 495     put.addColumn(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a"));
 496     put.addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("b"));
 497     put.addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c"));
 498     table.put(put).get();
 499
 500     // Put with success
 501     boolean ok = table.checkAndMutate(row, new FilterList(
 502         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"), CompareOperator.EQUAL,
 503           Bytes.toBytes("a")),
 504         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("B"), CompareOperator.EQUAL,
 505           Bytes.toBytes("b"))
 506       ))
 507       .thenPut(new Put(row).addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("d")))
 508       .get();
 509     assertTrue(ok);
 510
 511     Result result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("D"))).get();
 512     assertEquals("d", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("D"))));
 513
 514     // Put with failure
 515     ok = table.checkAndMutate(row, new FilterList(
 516         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"), CompareOperator.EQUAL,
 517           Bytes.toBytes("a")),
 518         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("B"), CompareOperator.EQUAL,
 519           Bytes.toBytes("c"))
 520       ))
 521       .thenPut(new Put(row).addColumn(FAMILY, Bytes.toBytes("E"), Bytes.toBytes("e")))
 522       .get();
 523     assertFalse(ok);
 524
 525     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("E"))).get());
 526
 527     // Delete with success
 528     ok = table.checkAndMutate(row, new FilterList(
 529         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"), CompareOperator.EQUAL,
 530           Bytes.toBytes("a")),
 531         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("B"), CompareOperator.EQUAL,
 532           Bytes.toBytes("b"))
 533       ))
 534       .thenDelete(new Delete(row).addColumns(FAMILY, Bytes.toBytes("D")))
 535       .get();
 536     assertTrue(ok);
 537
 538     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("D"))).get());
 539
 540     // Mutate with success
 541     ok = table.checkAndMutate(row, new FilterList(
 542         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"), CompareOperator.EQUAL,
 543           Bytes.toBytes("a")),
 544         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("B"), CompareOperator.EQUAL,
 545           Bytes.toBytes("b"))
 546       ))
 547       .thenMutate(new RowMutations(row)
 548         .add((Mutation) new Put(row)
 549           .addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("d")))
 550         .add((Mutation) new Delete(row).addColumns(FAMILY, Bytes.toBytes("A"))))
 551       .get();
 552     assertTrue(ok);
 553
 554     result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("D"))).get();
 555     assertEquals("d", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("D"))));
 556
 557     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("A"))).get());
 558   }
 559
 560   @Test
 561   @Deprecated
 562   public void testCheckAndMutateWithTimestampFilterForOldApi() throws Throwable {
 563     AsyncTable<?> table = getTable.get();
 564
 565     // Put with specifying the timestamp
 566     table.put(new Put(row).addColumn(FAMILY, Bytes.toBytes("A"), 100, Bytes.toBytes("a"))).get();
 567
 568     // Put with success
 569     boolean ok = table.checkAndMutate(row, new FilterList(
 570         new FamilyFilter(CompareOperator.EQUAL, new BinaryComparator(FAMILY)),
 571         new QualifierFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("A"))),
 572         new TimestampsFilter(Collections.singletonList(100L))
 573       ))
 574       .thenPut(new Put(row).addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("b")))
 575       .get();
 576     assertTrue(ok);
 577
 578     Result result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("B"))).get();
 579     assertEquals("b", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("B"))));
 580
 581     // Put with failure
 582     ok = table.checkAndMutate(row, new FilterList(
 583         new FamilyFilter(CompareOperator.EQUAL, new BinaryComparator(FAMILY)),
 584         new QualifierFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("A"))),
 585         new TimestampsFilter(Collections.singletonList(101L))
 586       ))
 587       .thenPut(new Put(row).addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c")))
 588       .get();
 589     assertFalse(ok);
 590
 591     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("C"))).get());
 592   }
 593
 594   @Test
 595   @Deprecated
 596   public void testCheckAndMutateWithFilterAndTimeRangeForOldApi() throws Throwable {
 597     AsyncTable<?> table = getTable.get();
 598
 599     // Put with specifying the timestamp
 600     table.put(new Put(row).addColumn(FAMILY, Bytes.toBytes("A"), 100, Bytes.toBytes("a")))
 601       .get();
 602
 603     // Put with success
 604     boolean ok = table.checkAndMutate(row, new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"),
 605         CompareOperator.EQUAL, Bytes.toBytes("a")))
 606       .timeRange(TimeRange.between(0, 101))
 607       .thenPut(new Put(row).addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("b")))
 608       .get();
 609     assertTrue(ok);
 610
 611     Result result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("B"))).get();
 612     assertEquals("b", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("B"))));
 613
 614     // Put with failure
 615     ok = table.checkAndMutate(row, new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"),
 616         CompareOperator.EQUAL, Bytes.toBytes("a")))
 617       .timeRange(TimeRange.between(0, 100))
 618       .thenPut(new Put(row).addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c")))
 619       .get();
 620     assertFalse(ok);
 621
 622     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("C"))).get());
 623   }
 624
 625   @Test(expected = NullPointerException.class)
 626   @Deprecated
 627   public void testCheckAndMutateWithoutConditionForOldApi() {
 628     getTable.get().checkAndMutate(row, FAMILY)
 629       .thenPut(new Put(row).addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("d")));
 630   }
 631
 632   // Tests for new CheckAndMutate API
 633
 634   @SuppressWarnings("FutureReturnValueIgnored")
 635   @Test
 636   public void testCheckAndPut() throws InterruptedException, ExecutionException {
 637     AsyncTable<?> table = getTable.get();
 638     AtomicInteger successCount = new AtomicInteger(0);
 639     AtomicInteger successIndex = new AtomicInteger(-1);
 640     int count = 10;
 641     CountDownLatch latch = new CountDownLatch(count);
 642
 643     IntStream.range(0, count)
 644       .forEach(i -> table.checkAndMutate(CheckAndMutate.newBuilder(row)
 645           .ifNotExists(FAMILY, QUALIFIER)
 646           .build(new Put(row).addColumn(FAMILY, QUALIFIER, concat(VALUE, i))))
 647         .thenAccept(x -> {
 648           if (x.isSuccess()) {
 649             successCount.incrementAndGet();
 650             successIndex.set(i);
 651           }
 652           assertNull(x.getResult());
 653           latch.countDown();
 654         }));
 655     latch.await();
 656     assertEquals(1, successCount.get());
 657     String actual = Bytes.toString(table.get(new Get(row)).get().getValue(FAMILY, QUALIFIER));
 658     assertTrue(actual.endsWith(Integer.toString(successIndex.get())));
 659   }
 660
 661   @SuppressWarnings("FutureReturnValueIgnored")
 662   @Test
 663   public void testCheckAndDelete() throws InterruptedException, ExecutionException {
 664     AsyncTable<?> table = getTable.get();
 665     int count = 10;
 666     CountDownLatch putLatch = new CountDownLatch(count + 1);
 667     table.put(new Put(row).addColumn(FAMILY, QUALIFIER, VALUE)).thenRun(() -> putLatch.countDown());
 668     IntStream.range(0, count)
 669       .forEach(i -> table.put(new Put(row).addColumn(FAMILY, concat(QUALIFIER, i), VALUE))
 670         .thenRun(() -> putLatch.countDown()));
 671     putLatch.await();
 672
 673     AtomicInteger successCount = new AtomicInteger(0);
 674     AtomicInteger successIndex = new AtomicInteger(-1);
 675     CountDownLatch deleteLatch = new CountDownLatch(count);
 676
 677     IntStream.range(0, count)
 678       .forEach(i -> table.checkAndMutate(CheckAndMutate.newBuilder(row)
 679           .ifEquals(FAMILY, QUALIFIER, VALUE)
 680           .build(
 681             new Delete(row).addColumn(FAMILY, QUALIFIER).addColumn(FAMILY, concat(QUALIFIER, i))))
 682         .thenAccept(x -> {
 683           if (x.isSuccess()) {
 684             successCount.incrementAndGet();
 685             successIndex.set(i);
 686           }
 687           assertNull(x.getResult());
 688           deleteLatch.countDown();
 689         }));
 690     deleteLatch.await();
 691     assertEquals(1, successCount.get());
 692     Result result = table.get(new Get(row)).get();
 693     IntStream.range(0, count).forEach(i -> {
 694       if (i == successIndex.get()) {
 695         assertFalse(result.containsColumn(FAMILY, concat(QUALIFIER, i)));
 696       } else {
 697         assertArrayEquals(VALUE, result.getValue(FAMILY, concat(QUALIFIER, i)));
 698       }
 699     });
 700   }
 701
 702   @SuppressWarnings("FutureReturnValueIgnored")
 703   @Test
 704   public void testCheckAndMutate() throws InterruptedException, ExecutionException {
 705     AsyncTable<?> table = getTable.get();
 706     int count = 10;
 707     CountDownLatch putLatch = new CountDownLatch(count + 1);
 708     table.put(new Put(row).addColumn(FAMILY, QUALIFIER, VALUE)).thenRun(() -> putLatch.countDown());
 709     IntStream.range(0, count)
 710       .forEach(i -> table.put(new Put(row).addColumn(FAMILY, concat(QUALIFIER, i), VALUE))
 711         .thenRun(() -> putLatch.countDown()));
 712     putLatch.await();
 713
 714     AtomicInteger successCount = new AtomicInteger(0);
 715     AtomicInteger successIndex = new AtomicInteger(-1);
 716     CountDownLatch mutateLatch = new CountDownLatch(count);
 717     IntStream.range(0, count).forEach(i -> {
 718       RowMutations mutation = new RowMutations(row);
 719       try {
 720         mutation.add((Mutation) new Delete(row).addColumn(FAMILY, QUALIFIER));
 721         mutation
 722           .add((Mutation) new Put(row).addColumn(FAMILY, concat(QUALIFIER, i), concat(VALUE, i)));
 723       } catch (IOException e) {
 724         throw new UncheckedIOException(e);
 725       }
 726
 727       table.checkAndMutate(CheckAndMutate.newBuilder(row)
 728           .ifEquals(FAMILY, QUALIFIER, VALUE)
 729           .build(mutation))
 730         .thenAccept(x -> {
 731           if (x.isSuccess()) {
 732             successCount.incrementAndGet();
 733             successIndex.set(i);
 734           }
 735           assertNull(x.getResult());
 736           mutateLatch.countDown();
 737         });
 738     });
 739     mutateLatch.await();
 740     assertEquals(1, successCount.get());
 741     Result result = table.get(new Get(row)).get();
 742     IntStream.range(0, count).forEach(i -> {
 743       if (i == successIndex.get()) {
 744         assertArrayEquals(concat(VALUE, i), result.getValue(FAMILY, concat(QUALIFIER, i)));
 745       } else {
 746         assertArrayEquals(VALUE, result.getValue(FAMILY, concat(QUALIFIER, i)));
 747       }
 748     });
 749   }
 750
 751   @Test
 752   public void testCheckAndMutateWithTimeRange() throws Exception {
 753     AsyncTable<?> table = getTable.get();
 754     final long ts = EnvironmentEdgeManager.currentTime() / 2;
 755     Put put = new Put(row);
 756     put.addColumn(FAMILY, QUALIFIER, ts, VALUE);
 757
 758     CheckAndMutateResult result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 759       .ifNotExists(FAMILY, QUALIFIER)
 760       .build(put)).get();
 761     assertTrue(result.isSuccess());
 762     assertNull(result.getResult());
 763
 764     result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 765       .ifEquals(FAMILY, QUALIFIER, VALUE)
 766       .timeRange(TimeRange.at(ts + 10000))
 767       .build(put)).get();
 768     assertFalse(result.isSuccess());
 769     assertNull(result.getResult());
 770
 771     result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 772       .ifEquals(FAMILY, QUALIFIER, VALUE)
 773       .timeRange(TimeRange.at(ts))
 774       .build(put)).get();
 775     assertTrue(result.isSuccess());
 776     assertNull(result.getResult());
 777
 778     RowMutations rm = new RowMutations(row).add((Mutation) put);
 779
 780     result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 781       .ifEquals(FAMILY, QUALIFIER, VALUE)
 782       .timeRange(TimeRange.at(ts + 10000))
 783       .build(rm)).get();
 784     assertFalse(result.isSuccess());
 785     assertNull(result.getResult());
 786
 787     result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 788       .ifEquals(FAMILY, QUALIFIER, VALUE)
 789       .timeRange(TimeRange.at(ts))
 790       .build(rm)).get();
 791     assertTrue(result.isSuccess());
 792     assertNull(result.getResult());
 793
 794     Delete delete = new Delete(row).addColumn(FAMILY, QUALIFIER);
 795
 796     result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 797       .ifEquals(FAMILY, QUALIFIER, VALUE)
 798       .timeRange(TimeRange.at(ts + 10000))
 799       .build(delete)).get();
 800     assertFalse(result.isSuccess());
 801     assertNull(result.getResult());
 802
 803     result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 804       .ifEquals(FAMILY, QUALIFIER, VALUE)
 805       .timeRange(TimeRange.at(ts))
 806       .build(delete)).get();
 807     assertTrue(result.isSuccess());
 808     assertNull(result.getResult());
 809   }
 810
 811   @Test
 812   public void testCheckAndMutateWithSingleFilter() throws Throwable {
 813     AsyncTable<?> table = getTable.get();
 814
 815     // Put one row
 816     Put put = new Put(row);
 817     put.addColumn(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a"));
 818     put.addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("b"));
 819     put.addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c"));
 820     table.put(put).get();
 821
 822     // Put with success
 823     CheckAndMutateResult result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 824       .ifMatches(new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"),
 825         CompareOperator.EQUAL, Bytes.toBytes("a")))
 826       .build(new Put(row).addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("d")))).get();
 827     assertTrue(result.isSuccess());
 828     assertNull(result.getResult());
 829
 830     Result r = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("D"))).get();
 831     assertEquals("d", Bytes.toString(r.getValue(FAMILY, Bytes.toBytes("D"))));
 832
 833     // Put with failure
 834     result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 835       .ifMatches(new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"),
 836         CompareOperator.EQUAL, Bytes.toBytes("b")))
 837       .build(new Put(row).addColumn(FAMILY, Bytes.toBytes("E"), Bytes.toBytes("e")))).get();
 838     assertFalse(result.isSuccess());
 839     assertNull(result.getResult());
 840
 841     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("E"))).get());
 842
 843     // Delete with success
 844     result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 845       .ifMatches(new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"),
 846         CompareOperator.EQUAL, Bytes.toBytes("a")))
 847       .build(new Delete(row).addColumns(FAMILY, Bytes.toBytes("D")))).get();
 848     assertTrue(result.isSuccess());
 849     assertNull(result.getResult());
 850
 851     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("D"))).get());
 852
 853     // Mutate with success
 854     result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 855       .ifMatches(new SingleColumnValueFilter(FAMILY, Bytes.toBytes("B"),
 856         CompareOperator.EQUAL, Bytes.toBytes("b")))
 857       .build(new RowMutations(row)
 858         .add((Mutation) new Put(row)
 859           .addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("d")))
 860         .add((Mutation) new Delete(row).addColumns(FAMILY, Bytes.toBytes("A"))))).get();
 861     assertTrue(result.isSuccess());
 862     assertNull(result.getResult());
 863
 864     r = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("D"))).get();
 865     assertEquals("d", Bytes.toString(r.getValue(FAMILY, Bytes.toBytes("D"))));
 866
 867     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("A"))).get());
 868   }
 869
 870   @Test
 871   public void testCheckAndMutateWithMultipleFilters() throws Throwable {
 872     AsyncTable<?> table = getTable.get();
 873
 874     // Put one row
 875     Put put = new Put(row);
 876     put.addColumn(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a"));
 877     put.addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("b"));
 878     put.addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c"));
 879     table.put(put).get();
 880
 881     // Put with success
 882     CheckAndMutateResult result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 883       .ifMatches(new FilterList(
 884         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"), CompareOperator.EQUAL,
 885           Bytes.toBytes("a")),
 886         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("B"), CompareOperator.EQUAL,
 887           Bytes.toBytes("b"))))
 888       .build(new Put(row).addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("d")))).get();
 889     assertTrue(result.isSuccess());
 890     assertNull(result.getResult());
 891
 892     Result r = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("D"))).get();
 893     assertEquals("d", Bytes.toString(r.getValue(FAMILY, Bytes.toBytes("D"))));
 894
 895     // Put with failure
 896     result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 897       .ifMatches(new FilterList(
 898         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"), CompareOperator.EQUAL,
 899           Bytes.toBytes("a")),
 900         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("B"), CompareOperator.EQUAL,
 901           Bytes.toBytes("c"))))
 902       .build(new Put(row).addColumn(FAMILY, Bytes.toBytes("E"), Bytes.toBytes("e")))).get();
 903     assertFalse(result.isSuccess());
 904     assertNull(result.getResult());
 905
 906     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("E"))).get());
 907
 908     // Delete with success
 909     result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 910       .ifMatches(new FilterList(
 911         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"), CompareOperator.EQUAL,
 912           Bytes.toBytes("a")),
 913         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("B"), CompareOperator.EQUAL,
 914           Bytes.toBytes("b"))))
 915       .build(new Delete(row).addColumns(FAMILY, Bytes.toBytes("D")))).get();
 916     assertTrue(result.isSuccess());
 917     assertNull(result.getResult());
 918
 919     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("D"))).get());
 920
 921     // Mutate with success
 922     result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 923       .ifMatches(new FilterList(
 924         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"), CompareOperator.EQUAL,
 925           Bytes.toBytes("a")),
 926         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("B"), CompareOperator.EQUAL,
 927           Bytes.toBytes("b"))))
 928       .build(new RowMutations(row)
 929         .add((Mutation) new Put(row)
 930           .addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("d")))
 931         .add((Mutation) new Delete(row).addColumns(FAMILY, Bytes.toBytes("A"))))).get();
 932     assertTrue(result.isSuccess());
 933     assertNull(result.getResult());
 934
 935     r = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("D"))).get();
 936     assertEquals("d", Bytes.toString(r.getValue(FAMILY, Bytes.toBytes("D"))));
 937
 938     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("A"))).get());
 939   }
 940
 941   @Test
 942   public void testCheckAndMutateWithTimestampFilter() throws Throwable {
 943     AsyncTable<?> table = getTable.get();
 944
 945     // Put with specifying the timestamp
 946     table.put(new Put(row).addColumn(FAMILY, Bytes.toBytes("A"), 100, Bytes.toBytes("a"))).get();
 947
 948     // Put with success
 949     CheckAndMutateResult result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 950       .ifMatches(new FilterList(
 951         new FamilyFilter(CompareOperator.EQUAL, new BinaryComparator(FAMILY)),
 952         new QualifierFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("A"))),
 953         new TimestampsFilter(Collections.singletonList(100L))))
 954       .build(new Put(row).addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("b")))).get();
 955     assertTrue(result.isSuccess());
 956     assertNull(result.getResult());
 957
 958     Result r = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("B"))).get();
 959     assertEquals("b", Bytes.toString(r.getValue(FAMILY, Bytes.toBytes("B"))));
 960
 961     // Put with failure
 962     result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 963       .ifMatches(new FilterList(
 964         new FamilyFilter(CompareOperator.EQUAL, new BinaryComparator(FAMILY)),
 965         new QualifierFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("A"))),
 966         new TimestampsFilter(Collections.singletonList(101L))))
 967       .build(new Put(row).addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c")))).get();
 968     assertFalse(result.isSuccess());
 969     assertNull(result.getResult());
 970
 971     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("C"))).get());
 972   }
 973
 974   @Test
 975   public void testCheckAndMutateWithFilterAndTimeRange() throws Throwable {
 976     AsyncTable<?> table = getTable.get();
 977
 978     // Put with specifying the timestamp
 979     table.put(new Put(row).addColumn(FAMILY, Bytes.toBytes("A"), 100, Bytes.toBytes("a")))
 980       .get();
 981
 982     // Put with success
 983     CheckAndMutateResult result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 984       .ifMatches(new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"),
 985         CompareOperator.EQUAL, Bytes.toBytes("a")))
 986       .timeRange(TimeRange.between(0, 101))
 987       .build(new Put(row).addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("b")))).get();
 988     assertTrue(result.isSuccess());
 989     assertNull(result.getResult());
 990
 991     Result r = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("B"))).get();
 992     assertEquals("b", Bytes.toString(r.getValue(FAMILY, Bytes.toBytes("B"))));
 993
 994     // Put with failure
 995     result = table.checkAndMutate(CheckAndMutate.newBuilder(row)
 996       .ifMatches(new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"),
 997         CompareOperator.EQUAL, Bytes.toBytes("a")))
 998       .timeRange(TimeRange.between(0, 100))
 999       .build(new Put(row).addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c"))))
1000       .get();
1001     assertFalse(result.isSuccess());
1002     assertNull(result.getResult());
1003
1004     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("C"))).get());
1005   }
1006
1007   @Test
1008   public void testCheckAndIncrement() throws Throwable {
1009     AsyncTable<?> table = getTable.get();
1010
1011     table.put(new Put(row).addColumn(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a"))).get();
1012
1013     // CheckAndIncrement with correct value
1014     CheckAndMutateResult res = table.checkAndMutate(CheckAndMutate.newBuilder(row)
1015       .ifEquals(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a"))
1016       .build(new Increment(row).addColumn(FAMILY, Bytes.toBytes("B"), 1))).get();
1017     assertTrue(res.isSuccess());
1018     assertEquals(1, Bytes.toLong(res.getResult().getValue(FAMILY, Bytes.toBytes("B"))));
1019
1020     Result result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("B"))).get();
1021     assertEquals(1, Bytes.toLong(result.getValue(FAMILY, Bytes.toBytes("B"))));
1022
1023     // CheckAndIncrement with wrong value
1024     res = table.checkAndMutate(CheckAndMutate.newBuilder(row)
1025       .ifEquals(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("b"))
1026       .build(new Increment(row).addColumn(FAMILY, Bytes.toBytes("B"), 1))).get();
1027     assertFalse(res.isSuccess());
1028     assertNull(res.getResult());
1029
1030     result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("B"))).get();
1031     assertEquals(1, Bytes.toLong(result.getValue(FAMILY, Bytes.toBytes("B"))));
1032
1033     table.put(new Put(row).addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c")));
1034
1035     // CheckAndIncrement with a filter and correct value
1036     res = table.checkAndMutate(CheckAndMutate.newBuilder(row)
1037       .ifMatches(new FilterList(
1038         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"), CompareOperator.EQUAL,
1039           Bytes.toBytes("a")),
1040         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("C"), CompareOperator.EQUAL,
1041           Bytes.toBytes("c"))))
1042       .build(new Increment(row).addColumn(FAMILY, Bytes.toBytes("B"), 2))).get();
1043     assertTrue(res.isSuccess());
1044     assertEquals(3, Bytes.toLong(res.getResult().getValue(FAMILY, Bytes.toBytes("B"))));
1045
1046     result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("B"))).get();
1047     assertEquals(3, Bytes.toLong(result.getValue(FAMILY, Bytes.toBytes("B"))));
1048
1049     // CheckAndIncrement with a filter and correct value
1050     res = table.checkAndMutate(CheckAndMutate.newBuilder(row)
1051       .ifMatches(new FilterList(
1052         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"), CompareOperator.EQUAL,
1053           Bytes.toBytes("b")),
1054         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("C"), CompareOperator.EQUAL,
1055           Bytes.toBytes("d"))))
1056       .build(new Increment(row).addColumn(FAMILY, Bytes.toBytes("B"), 2))).get();
1057     assertFalse(res.isSuccess());
1058     assertNull(res.getResult());
1059
1060     result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("B"))).get();
1061     assertEquals(3, Bytes.toLong(result.getValue(FAMILY, Bytes.toBytes("B"))));
1062   }
1063
1064   @Test
1065   public void testCheckAndAppend() throws Throwable {
1066     AsyncTable<?> table = getTable.get();
1067
1068     table.put(new Put(row).addColumn(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a"))).get();
1069
1070     // CheckAndAppend with correct value
1071     CheckAndMutateResult res = table.checkAndMutate(CheckAndMutate.newBuilder(row)
1072       .ifEquals(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a"))
1073       .build(new Append(row).addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("b")))).get();
1074     assertTrue(res.isSuccess());
1075     assertEquals("b", Bytes.toString(res.getResult().getValue(FAMILY, Bytes.toBytes("B"))));
1076
1077     Result result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("B"))).get();
1078     assertEquals("b", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("B"))));
1079
1080     // CheckAndAppend with correct value
1081     res = table.checkAndMutate(CheckAndMutate.newBuilder(row)
1082       .ifEquals(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("b"))
1083       .build(new Append(row).addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("b")))).get();
1084     assertFalse(res.isSuccess());
1085     assertNull(res.getResult());
1086
1087     result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("B"))).get();
1088     assertEquals("b", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("B"))));
1089
1090     table.put(new Put(row).addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c")));
1091
1092     // CheckAndAppend with a filter and correct value
1093     res = table.checkAndMutate(CheckAndMutate.newBuilder(row)
1094       .ifMatches(new FilterList(
1095         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"), CompareOperator.EQUAL,
1096           Bytes.toBytes("a")),
1097         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("C"), CompareOperator.EQUAL,
1098           Bytes.toBytes("c"))))
1099       .build(new Append(row).addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("bb")))).get();
1100     assertTrue(res.isSuccess());
1101     assertEquals("bbb", Bytes.toString(res.getResult().getValue(FAMILY, Bytes.toBytes("B"))));
1102
1103     result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("B"))).get();
1104     assertEquals("bbb", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("B"))));
1105
1106     // CheckAndAppend with a filter and wrong value
1107     res = table.checkAndMutate(CheckAndMutate.newBuilder(row)
1108       .ifMatches(new FilterList(
1109         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"), CompareOperator.EQUAL,
1110           Bytes.toBytes("b")),
1111         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("C"), CompareOperator.EQUAL,
1112           Bytes.toBytes("d"))))
1113       .build(new Append(row).addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("bb")))).get();
1114     assertFalse(res.isSuccess());
1115     assertNull(res.getResult());
1116
1117     result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("B"))).get();
1118     assertEquals("bbb", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("B"))));
1119   }
1120
1121   @Test
1122   public void testCheckAndRowMutations() throws Throwable {
1123     final byte[] q1 = Bytes.toBytes("q1");
1124     final byte[] q2 = Bytes.toBytes("q2");
1125     final byte[] q3 = Bytes.toBytes("q3");
1126     final byte[] q4 = Bytes.toBytes("q4");
1127     final String v1 = "v1";
1128
1129     AsyncTable<?> table = getTable.get();
1130
1131     // Initial values
1132     table.putAll(Arrays.asList(
1133       new Put(row).addColumn(FAMILY, q2, Bytes.toBytes("toBeDeleted")),
1134       new Put(row).addColumn(FAMILY, q3, Bytes.toBytes(5L)),
1135       new Put(row).addColumn(FAMILY, q4, Bytes.toBytes("a")))).get();
1136
1137     // Do CheckAndRowMutations
1138     CheckAndMutate checkAndMutate = CheckAndMutate.newBuilder(row)
1139       .ifNotExists(FAMILY, q1)
1140       .build(new RowMutations(row).add(Arrays.asList(
1141         new Put(row).addColumn(FAMILY, q1, Bytes.toBytes(v1)),
1142         new Delete(row).addColumns(FAMILY, q2),
1143         new Increment(row).addColumn(FAMILY, q3, 1),
1144         new Append(row).addColumn(FAMILY, q4, Bytes.toBytes("b"))))
1145       );
1146
1147     CheckAndMutateResult result = table.checkAndMutate(checkAndMutate).get();
1148     assertTrue(result.isSuccess());
1149     assertEquals(6L, Bytes.toLong(result.getResult().getValue(FAMILY, q3)));
1150     assertEquals("ab", Bytes.toString(result.getResult().getValue(FAMILY, q4)));
1151
1152     // Verify the value
1153     Result r = table.get(new Get(row)).get();
1154     assertEquals(v1, Bytes.toString(r.getValue(FAMILY, q1)));
1155     assertNull(r.getValue(FAMILY, q2));
1156     assertEquals(6L, Bytes.toLong(r.getValue(FAMILY, q3)));
1157     assertEquals("ab", Bytes.toString(r.getValue(FAMILY, q4)));
1158
1159     // Do CheckAndRowMutations again
1160     checkAndMutate = CheckAndMutate.newBuilder(row)
1161       .ifNotExists(FAMILY, q1)
1162       .build(new RowMutations(row).add(Arrays.asList(
1163         new Delete(row).addColumns(FAMILY, q1),
1164         new Put(row).addColumn(FAMILY, q2, Bytes.toBytes(v1)),
1165         new Increment(row).addColumn(FAMILY, q3, 1),
1166         new Append(row).addColumn(FAMILY, q4, Bytes.toBytes("b"))))
1167       );
1168
1169     result = table.checkAndMutate(checkAndMutate).get();
1170     assertFalse(result.isSuccess());
1171     assertNull(result.getResult());
1172
1173     // Verify the value
1174     r = table.get(new Get(row)).get();
1175     assertEquals(v1, Bytes.toString(r.getValue(FAMILY, q1)));
1176     assertNull(r.getValue(FAMILY, q2));
1177     assertEquals(6L, Bytes.toLong(r.getValue(FAMILY, q3)));
1178     assertEquals("ab", Bytes.toString(r.getValue(FAMILY, q4)));
1179   }
1180
1181   // Tests for batch version of checkAndMutate
1182
1183   @Test
1184   public void testCheckAndMutateBatch() throws Throwable {
1185     AsyncTable<?> table = getTable.get();
1186     byte[] row2 = Bytes.toBytes(Bytes.toString(row) + "2");
1187     byte[] row3 = Bytes.toBytes(Bytes.toString(row) + "3");
1188     byte[] row4 = Bytes.toBytes(Bytes.toString(row) + "4");
1189
1190     table.putAll(Arrays.asList(
1191       new Put(row).addColumn(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a")),
1192       new Put(row2).addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("b")),
1193       new Put(row3).addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c")),
1194       new Put(row4).addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("d")))).get();
1195
1196     // Test for Put
1197     CheckAndMutate checkAndMutate1 = CheckAndMutate.newBuilder(row)
1198       .ifEquals(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a"))
1199       .build(new Put(row).addColumn(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("e")));
1200
1201     CheckAndMutate checkAndMutate2 = CheckAndMutate.newBuilder(row2)
1202       .ifEquals(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("a"))
1203       .build(new Put(row2).addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("f")));
1204
1205     List<CheckAndMutateResult> results =
1206       table.checkAndMutateAll(Arrays.asList(checkAndMutate1, checkAndMutate2)).get();
1207
1208     assertTrue(results.get(0).isSuccess());
1209     assertNull(results.get(0).getResult());
1210     assertFalse(results.get(1).isSuccess());
1211     assertNull(results.get(1).getResult());
1212
1213     Result result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("A"))).get();
1214     assertEquals("e", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("A"))));
1215
1216     result = table.get(new Get(row2).addColumn(FAMILY, Bytes.toBytes("B"))).get();
1217     assertEquals("b", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("B"))));
1218
1219     // Test for Delete
1220     checkAndMutate1 = CheckAndMutate.newBuilder(row)
1221       .ifEquals(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("e"))
1222       .build(new Delete(row));
1223
1224     checkAndMutate2 = CheckAndMutate.newBuilder(row2)
1225       .ifEquals(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("a"))
1226       .build(new Delete(row2));
1227
1228     results = table.checkAndMutateAll(Arrays.asList(checkAndMutate1, checkAndMutate2)).get();
1229
1230     assertTrue(results.get(0).isSuccess());
1231     assertNull(results.get(0).getResult());
1232     assertFalse(results.get(1).isSuccess());
1233     assertNull(results.get(1).getResult());
1234
1235     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("A"))).get());
1236
1237     result = table.get(new Get(row2).addColumn(FAMILY, Bytes.toBytes("B"))).get();
1238     assertEquals("b", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("B"))));
1239
1240     // Test for RowMutations
1241     checkAndMutate1 = CheckAndMutate.newBuilder(row3)
1242       .ifEquals(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c"))
1243       .build(new RowMutations(row3)
1244         .add((Mutation) new Put(row3)
1245           .addColumn(FAMILY, Bytes.toBytes("F"), Bytes.toBytes("f")))
1246         .add((Mutation) new Delete(row3).addColumns(FAMILY, Bytes.toBytes("C"))));
1247
1248     checkAndMutate2 = CheckAndMutate.newBuilder(row4)
1249       .ifEquals(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("f"))
1250       .build(new RowMutations(row4)
1251         .add((Mutation) new Put(row4)
1252           .addColumn(FAMILY, Bytes.toBytes("F"), Bytes.toBytes("f")))
1253         .add((Mutation) new Delete(row4).addColumns(FAMILY, Bytes.toBytes("D"))));
1254
1255     results = table.checkAndMutateAll(Arrays.asList(checkAndMutate1, checkAndMutate2)).get();
1256
1257     assertTrue(results.get(0).isSuccess());
1258     assertNull(results.get(0).getResult());
1259     assertFalse(results.get(1).isSuccess());
1260     assertNull(results.get(1).getResult());
1261
1262     result = table.get(new Get(row3)).get();
1263     assertEquals("f", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("F"))));
1264     assertNull(result.getValue(FAMILY, Bytes.toBytes("D")));
1265
1266     result = table.get(new Get(row4)).get();
1267     assertNull(result.getValue(FAMILY, Bytes.toBytes("F")));
1268     assertEquals("d", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("D"))));
1269   }
1270
1271   @Test
1272   public void testCheckAndMutateBatch2() throws Throwable {
1273     AsyncTable<?> table = getTable.get();
1274     byte[] row2 = Bytes.toBytes(Bytes.toString(row) + "2");
1275     byte[] row3 = Bytes.toBytes(Bytes.toString(row) + "3");
1276     byte[] row4 = Bytes.toBytes(Bytes.toString(row) + "4");
1277
1278     table.putAll(Arrays.asList(
1279       new Put(row).addColumn(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a")),
1280       new Put(row2).addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("b")),
1281       new Put(row3).addColumn(FAMILY, Bytes.toBytes("C"), 100, Bytes.toBytes("c")),
1282       new Put(row4).addColumn(FAMILY, Bytes.toBytes("D"), 100, Bytes.toBytes("d")))).get();
1283
1284     // Test for ifNotExists()
1285     CheckAndMutate checkAndMutate1 = CheckAndMutate.newBuilder(row)
1286       .ifNotExists(FAMILY, Bytes.toBytes("B"))
1287       .build(new Put(row).addColumn(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("e")));
1288
1289     CheckAndMutate checkAndMutate2 = CheckAndMutate.newBuilder(row2)
1290       .ifNotExists(FAMILY, Bytes.toBytes("B"))
1291       .build(new Put(row2).addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("f")));
1292
1293     List<CheckAndMutateResult> results =
1294       table.checkAndMutateAll(Arrays.asList(checkAndMutate1, checkAndMutate2)).get();
1295
1296     assertTrue(results.get(0).isSuccess());
1297     assertNull(results.get(0).getResult());
1298     assertFalse(results.get(1).isSuccess());
1299     assertNull(results.get(1).getResult());
1300
1301     Result result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("A"))).get();
1302     assertEquals("e", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("A"))));
1303
1304     result = table.get(new Get(row2).addColumn(FAMILY, Bytes.toBytes("B"))).get();
1305     assertEquals("b", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("B"))));
1306
1307     // Test for ifMatches()
1308     checkAndMutate1 = CheckAndMutate.newBuilder(row)
1309       .ifMatches(FAMILY, Bytes.toBytes("A"), CompareOperator.NOT_EQUAL, Bytes.toBytes("a"))
1310       .build(new Put(row).addColumn(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a")));
1311
1312     checkAndMutate2 = CheckAndMutate.newBuilder(row2)
1313       .ifMatches(FAMILY, Bytes.toBytes("B"), CompareOperator.GREATER, Bytes.toBytes("b"))
1314       .build(new Put(row2).addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("f")));
1315
1316     results = table.checkAndMutateAll(Arrays.asList(checkAndMutate1, checkAndMutate2)).get();
1317
1318     assertTrue(results.get(0).isSuccess());
1319     assertNull(results.get(0).getResult());
1320     assertFalse(results.get(1).isSuccess());
1321     assertNull(results.get(1).getResult());
1322
1323     result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("A"))).get();
1324     assertEquals("a", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("A"))));
1325
1326     result = table.get(new Get(row2).addColumn(FAMILY, Bytes.toBytes("B"))).get();
1327     assertEquals("b", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("B"))));
1328
1329     // Test for timeRange()
1330     checkAndMutate1 = CheckAndMutate.newBuilder(row3)
1331       .ifEquals(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c"))
1332       .timeRange(TimeRange.between(0, 101))
1333       .build(new Put(row3).addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("e")));
1334
1335     checkAndMutate2 = CheckAndMutate.newBuilder(row4)
1336       .ifEquals(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("d"))
1337       .timeRange(TimeRange.between(0, 100))
1338       .build(new Put(row4).addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("f")));
1339
1340     results = table.checkAndMutateAll(Arrays.asList(checkAndMutate1, checkAndMutate2)).get();
1341
1342     assertTrue(results.get(0).isSuccess());
1343     assertNull(results.get(0).getResult());
1344     assertFalse(results.get(1).isSuccess());
1345     assertNull(results.get(1).getResult());
1346
1347     result = table.get(new Get(row3).addColumn(FAMILY, Bytes.toBytes("C"))).get();
1348     assertEquals("e", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("C"))));
1349
1350     result = table.get(new Get(row4).addColumn(FAMILY, Bytes.toBytes("D"))).get();
1351     assertEquals("d", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("D"))));
1352   }
1353
1354   @Test
1355   public void testCheckAndMutateBatchWithFilter() throws Throwable {
1356     AsyncTable<?> table = getTable.get();
1357     byte[] row2 = Bytes.toBytes(Bytes.toString(row) + "2");
1358
1359     table.putAll(Arrays.asList(
1360       new Put(row)
1361         .addColumn(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a"))
1362         .addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("b"))
1363         .addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c")),
1364       new Put(row2)
1365         .addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("d"))
1366         .addColumn(FAMILY, Bytes.toBytes("E"), Bytes.toBytes("e"))
1367         .addColumn(FAMILY, Bytes.toBytes("F"), Bytes.toBytes("f")))).get();
1368
1369     // Test for Put
1370     CheckAndMutate checkAndMutate1 = CheckAndMutate.newBuilder(row)
1371       .ifMatches(new FilterList(
1372         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"), CompareOperator.EQUAL,
1373           Bytes.toBytes("a")),
1374         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("B"), CompareOperator.EQUAL,
1375           Bytes.toBytes("b"))))
1376       .build(new Put(row).addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("g")));
1377
1378     CheckAndMutate checkAndMutate2 = CheckAndMutate.newBuilder(row2)
1379       .ifMatches(new FilterList(
1380         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("D"), CompareOperator.EQUAL,
1381           Bytes.toBytes("a")),
1382         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("E"), CompareOperator.EQUAL,
1383           Bytes.toBytes("b"))))
1384       .build(new Put(row2).addColumn(FAMILY, Bytes.toBytes("F"), Bytes.toBytes("h")));
1385
1386     List<CheckAndMutateResult> results =
1387       table.checkAndMutateAll(Arrays.asList(checkAndMutate1, checkAndMutate2)).get();
1388
1389     assertTrue(results.get(0).isSuccess());
1390     assertNull(results.get(0).getResult());
1391     assertFalse(results.get(1).isSuccess());
1392     assertNull(results.get(1).getResult());
1393
1394     Result result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("C"))).get();
1395     assertEquals("g", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("C"))));
1396
1397     result = table.get(new Get(row2).addColumn(FAMILY, Bytes.toBytes("F"))).get();
1398     assertEquals("f", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("F"))));
1399
1400     // Test for Delete
1401     checkAndMutate1 = CheckAndMutate.newBuilder(row)
1402       .ifMatches(new FilterList(
1403         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"), CompareOperator.EQUAL,
1404           Bytes.toBytes("a")),
1405         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("B"), CompareOperator.EQUAL,
1406           Bytes.toBytes("b"))))
1407       .build(new Delete(row).addColumns(FAMILY, Bytes.toBytes("C")));
1408
1409     checkAndMutate2 = CheckAndMutate.newBuilder(row2)
1410       .ifMatches(new FilterList(
1411         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("D"), CompareOperator.EQUAL,
1412           Bytes.toBytes("a")),
1413         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("E"), CompareOperator.EQUAL,
1414           Bytes.toBytes("b"))))
1415       .build(new Delete(row2).addColumn(FAMILY, Bytes.toBytes("F")));
1416
1417     results = table.checkAndMutateAll(Arrays.asList(checkAndMutate1, checkAndMutate2)).get();
1418
1419     assertTrue(results.get(0).isSuccess());
1420     assertNull(results.get(0).getResult());
1421     assertFalse(results.get(1).isSuccess());
1422     assertNull(results.get(1).getResult());
1423
1424     assertFalse(table.exists(new Get(row).addColumn(FAMILY, Bytes.toBytes("C"))).get());
1425
1426     result = table.get(new Get(row2).addColumn(FAMILY, Bytes.toBytes("F"))).get();
1427     assertEquals("f", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("F"))));
1428
1429     // Test for RowMutations
1430     checkAndMutate1 = CheckAndMutate.newBuilder(row)
1431       .ifMatches(new FilterList(
1432         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"), CompareOperator.EQUAL,
1433           Bytes.toBytes("a")),
1434         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("B"), CompareOperator.EQUAL,
1435           Bytes.toBytes("b"))))
1436       .build(new RowMutations(row)
1437         .add((Mutation) new Put(row)
1438           .addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c")))
1439         .add((Mutation) new Delete(row).addColumns(FAMILY, Bytes.toBytes("A"))));
1440
1441     checkAndMutate2 = CheckAndMutate.newBuilder(row2)
1442       .ifMatches(new FilterList(
1443         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("D"), CompareOperator.EQUAL,
1444           Bytes.toBytes("a")),
1445         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("E"), CompareOperator.EQUAL,
1446           Bytes.toBytes("b"))))
1447       .build(new RowMutations(row2)
1448         .add((Mutation) new Put(row2)
1449           .addColumn(FAMILY, Bytes.toBytes("F"), Bytes.toBytes("g")))
1450         .add((Mutation) new Delete(row2).addColumns(FAMILY, Bytes.toBytes("D"))));
1451
1452     results = table.checkAndMutateAll(Arrays.asList(checkAndMutate1, checkAndMutate2)).get();
1453
1454     assertTrue(results.get(0).isSuccess());
1455     assertNull(results.get(0).getResult());
1456     assertFalse(results.get(1).isSuccess());
1457     assertNull(results.get(1).getResult());
1458
1459     result = table.get(new Get(row)).get();
1460     assertNull(result.getValue(FAMILY, Bytes.toBytes("A")));
1461     assertEquals("c", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("C"))));
1462
1463     result = table.get(new Get(row2)).get();
1464     assertEquals("d", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("D"))));
1465     assertEquals("f", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("F"))));
1466   }
1467
1468   @Test
1469   public void testCheckAndMutateBatchWithFilterAndTimeRange() throws Throwable {
1470     AsyncTable<?> table = getTable.get();
1471     byte[] row2 = Bytes.toBytes(Bytes.toString(row) + "2");
1472
1473     table.putAll(Arrays.asList(
1474       new Put(row).addColumn(FAMILY, Bytes.toBytes("A"), 100, Bytes.toBytes("a"))
1475         .addColumn(FAMILY, Bytes.toBytes("B"), 100, Bytes.toBytes("b"))
1476         .addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c")),
1477       new Put(row2).addColumn(FAMILY, Bytes.toBytes("D"), 100, Bytes.toBytes("d"))
1478         .addColumn(FAMILY, Bytes.toBytes("E"), 100, Bytes.toBytes("e"))
1479         .addColumn(FAMILY, Bytes.toBytes("F"), Bytes.toBytes("f")))).get();
1480
1481     CheckAndMutate checkAndMutate1 = CheckAndMutate.newBuilder(row)
1482       .ifMatches(new FilterList(
1483         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("A"), CompareOperator.EQUAL,
1484           Bytes.toBytes("a")),
1485         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("B"), CompareOperator.EQUAL,
1486           Bytes.toBytes("b"))))
1487       .timeRange(TimeRange.between(0, 101))
1488       .build(new Put(row).addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("g")));
1489
1490     CheckAndMutate checkAndMutate2 = CheckAndMutate.newBuilder(row2)
1491       .ifMatches(new FilterList(
1492         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("D"), CompareOperator.EQUAL,
1493           Bytes.toBytes("d")),
1494         new SingleColumnValueFilter(FAMILY, Bytes.toBytes("E"), CompareOperator.EQUAL,
1495           Bytes.toBytes("e"))))
1496       .timeRange(TimeRange.between(0, 100))
1497       .build(new Put(row2).addColumn(FAMILY, Bytes.toBytes("F"), Bytes.toBytes("h")));
1498
1499     List<CheckAndMutateResult> results =
1500       table.checkAndMutateAll(Arrays.asList(checkAndMutate1, checkAndMutate2)).get();
1501
1502     assertTrue(results.get(0).isSuccess());
1503     assertNull(results.get(0).getResult());
1504     assertFalse(results.get(1).isSuccess());
1505     assertNull(results.get(1).getResult());
1506
1507     Result result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("C"))).get();
1508     assertEquals("g", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("C"))));
1509
1510     result = table.get(new Get(row2).addColumn(FAMILY, Bytes.toBytes("F"))).get();
1511     assertEquals("f", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("F"))));
1512   }
1513
1514   @Test
1515   public void testCheckAndIncrementBatch() throws Throwable {
1516     AsyncTable<?> table = getTable.get();
1517     byte[] row2 = Bytes.toBytes(Bytes.toString(row) + "2");
1518
1519     table.putAll(Arrays.asList(
1520       new Put(row).addColumn(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a"))
1521         .addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes(0L)),
1522       new Put(row2).addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c"))
1523         .addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes(0L)))).get();
1524
1525     // CheckAndIncrement with correct value
1526     CheckAndMutate checkAndMutate1 = CheckAndMutate.newBuilder(row)
1527       .ifEquals(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a"))
1528       .build(new Increment(row).addColumn(FAMILY, Bytes.toBytes("B"), 1));
1529
1530     // CheckAndIncrement with wrong value
1531     CheckAndMutate checkAndMutate2 = CheckAndMutate.newBuilder(row2)
1532       .ifEquals(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("d"))
1533       .build(new Increment(row2).addColumn(FAMILY, Bytes.toBytes("D"), 1));
1534
1535     List<CheckAndMutateResult> results =
1536       table.checkAndMutateAll(Arrays.asList(checkAndMutate1, checkAndMutate2)).get();
1537
1538     assertTrue(results.get(0).isSuccess());
1539     assertEquals(1, Bytes.toLong(results.get(0).getResult()
1540       .getValue(FAMILY, Bytes.toBytes("B"))));
1541     assertFalse(results.get(1).isSuccess());
1542     assertNull(results.get(1).getResult());
1543
1544     Result result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("B"))).get();
1545     assertEquals(1, Bytes.toLong(result.getValue(FAMILY, Bytes.toBytes("B"))));
1546
1547     result = table.get(new Get(row2).addColumn(FAMILY, Bytes.toBytes("D"))).get();
1548     assertEquals(0, Bytes.toLong(result.getValue(FAMILY, Bytes.toBytes("D"))));
1549   }
1550
1551   @Test
1552   public void testCheckAndAppendBatch() throws Throwable {
1553     AsyncTable<?> table = getTable.get();
1554     byte[] row2 = Bytes.toBytes(Bytes.toString(row) + "2");
1555
1556     table.putAll(Arrays.asList(
1557       new Put(row).addColumn(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a"))
1558         .addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("b")),
1559       new Put(row2).addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("c"))
1560         .addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("d")))).get();
1561
1562     // CheckAndAppend with correct value
1563     CheckAndMutate checkAndMutate1 = CheckAndMutate.newBuilder(row)
1564       .ifEquals(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a"))
1565       .build(new Append(row).addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("b")));
1566
1567     // CheckAndAppend with wrong value
1568     CheckAndMutate checkAndMutate2 = CheckAndMutate.newBuilder(row2)
1569       .ifEquals(FAMILY, Bytes.toBytes("C"), Bytes.toBytes("d"))
1570       .build(new Append(row2).addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("d")));
1571
1572     List<CheckAndMutateResult> results =
1573       table.checkAndMutateAll(Arrays.asList(checkAndMutate1, checkAndMutate2)).get();
1574
1575     assertTrue(results.get(0).isSuccess());
1576     assertEquals("bb", Bytes.toString(results.get(0).getResult()
1577       .getValue(FAMILY, Bytes.toBytes("B"))));
1578     assertFalse(results.get(1).isSuccess());
1579     assertNull(results.get(1).getResult());
1580
1581     Result result = table.get(new Get(row).addColumn(FAMILY, Bytes.toBytes("B"))).get();
1582     assertEquals("bb", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("B"))));
1583
1584     result = table.get(new Get(row2).addColumn(FAMILY, Bytes.toBytes("D"))).get();
1585     assertEquals("d", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("D"))));
1586   }
1587
1588   @Test
1589   public void testCheckAndRowMutationsBatch() throws Throwable {
1590     AsyncTable<?> table = getTable.get();
1591     byte[] row2 = Bytes.toBytes(Bytes.toString(row) + "2");
1592
1593     table.putAll(Arrays.asList(
1594       new Put(row).addColumn(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("b"))
1595         .addColumn(FAMILY, Bytes.toBytes("C"), Bytes.toBytes(1L))
1596         .addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("d")),
1597       new Put(row2).addColumn(FAMILY, Bytes.toBytes("F"), Bytes.toBytes("f"))
1598         .addColumn(FAMILY, Bytes.toBytes("G"), Bytes.toBytes(1L))
1599         .addColumn(FAMILY, Bytes.toBytes("H"), Bytes.toBytes("h")))
1600     ).get();
1601
1602     // CheckAndIncrement with correct value
1603     CheckAndMutate checkAndMutate1 = CheckAndMutate.newBuilder(row)
1604       .ifEquals(FAMILY, Bytes.toBytes("B"), Bytes.toBytes("b"))
1605       .build(new RowMutations(row).add(Arrays.asList(
1606         new Put(row).addColumn(FAMILY, Bytes.toBytes("A"), Bytes.toBytes("a")),
1607         new Delete(row).addColumns(FAMILY, Bytes.toBytes("B")),
1608         new Increment(row).addColumn(FAMILY, Bytes.toBytes("C"), 1L),
1609         new Append(row).addColumn(FAMILY, Bytes.toBytes("D"), Bytes.toBytes("d"))
1610       )));
1611
1612     // CheckAndIncrement with wrong value
1613     CheckAndMutate checkAndMutate2 = CheckAndMutate.newBuilder(row2)
1614       .ifEquals(FAMILY, Bytes.toBytes("F"), Bytes.toBytes("a"))
1615       .build(new RowMutations(row2).add(Arrays.asList(
1616         new Put(row2).addColumn(FAMILY, Bytes.toBytes("E"), Bytes.toBytes("e")),
1617         new Delete(row2).addColumns(FAMILY, Bytes.toBytes("F")),
1618         new Increment(row2).addColumn(FAMILY, Bytes.toBytes("G"), 1L),
1619         new Append(row2).addColumn(FAMILY, Bytes.toBytes("H"), Bytes.toBytes("h"))
1620       )));
1621
1622     List<CheckAndMutateResult> results =
1623       table.checkAndMutateAll(Arrays.asList(checkAndMutate1, checkAndMutate2)).get();
1624
1625     assertTrue(results.get(0).isSuccess());
1626     assertEquals(2, Bytes.toLong(results.get(0).getResult()
1627       .getValue(FAMILY, Bytes.toBytes("C"))));
1628     assertEquals("dd", Bytes.toString(results.get(0).getResult()
1629       .getValue(FAMILY, Bytes.toBytes("D"))));
1630
1631     assertFalse(results.get(1).isSuccess());
1632     assertNull(results.get(1).getResult());
1633
1634     Result result = table.get(new Get(row)).get();
1635     assertEquals("a", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("A"))));
1636     assertNull(result.getValue(FAMILY, Bytes.toBytes("B")));
1637     assertEquals(2, Bytes.toLong(result.getValue(FAMILY, Bytes.toBytes("C"))));
1638     assertEquals("dd", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("D"))));
1639
1640     result = table.get(new Get(row2)).get();
1641     assertNull(result.getValue(FAMILY, Bytes.toBytes("E")));
1642     assertEquals("f", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("F"))));
1643     assertEquals(1, Bytes.toLong(result.getValue(FAMILY, Bytes.toBytes("G"))));
1644     assertEquals("h", Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("H"))));
1645   }
1646
1647   @Test
1648   public void testDisabled() throws InterruptedException, ExecutionException {
1649     ASYNC_CONN.getAdmin().disableTable(TABLE_NAME).get();
1650     try {
1651       getTable.get().get(new Get(row)).get();
1652       fail("Should fail since table has been disabled");
1653     } catch (ExecutionException e) {
1654       Throwable cause = e.getCause();
1655       assertThat(cause, instanceOf(TableNotEnabledException.class));
1656       assertThat(cause.getMessage(), containsString(TABLE_NAME.getNameAsString()));
1657     }
1658   }
1659
1660   @Test
1661   public void testInvalidPut() {
1662     try {
1663       getTable.get().put(new Put(Bytes.toBytes(0)));
1664       fail("Should fail since the put does not contain any cells");
1665     } catch (IllegalArgumentException e) {
1666       assertThat(e.getMessage(), containsString("No columns to insert"));
1667     }
1668
1669     try {
1670       getTable.get()
1671         .put(new Put(Bytes.toBytes(0)).addColumn(FAMILY, QUALIFIER, new byte[MAX_KEY_VALUE_SIZE]));
1672       fail("Should fail since the put exceeds the max key value size");
1673     } catch (IllegalArgumentException e) {
1674       assertThat(e.getMessage(), containsString("KeyValue size too large"));
1675     }
1676   }
1677
1678   @Test
1679   public void testInvalidPutInRowMutations() throws IOException {
1680     final byte[] row = Bytes.toBytes(0);
1681     try {
1682       getTable.get().mutateRow(new RowMutations(row).add(new Put(row)));
1683       fail("Should fail since the put does not contain any cells");
1684     } catch (IllegalArgumentException e) {
1685       assertThat(e.getMessage(), containsString("No columns to insert"));
1686     }
1687
1688     try {
1689       getTable.get()
1690         .mutateRow(new RowMutations(row).add(new Put(row)
1691           .addColumn(FAMILY, QUALIFIER, new byte[MAX_KEY_VALUE_SIZE])));
1692       fail("Should fail since the put exceeds the max key value size");
1693     } catch (IllegalArgumentException e) {
1694       assertThat(e.getMessage(), containsString("KeyValue size too large"));
1695     }
1696   }
1697
1698   @Test
1699   public void testInvalidPutInRowMutationsInCheckAndMutate() throws IOException {
1700     final byte[] row = Bytes.toBytes(0);
1701     try {
1702       getTable.get().checkAndMutate(CheckAndMutate.newBuilder(row)
1703         .ifNotExists(FAMILY, QUALIFIER)
1704         .build(new RowMutations(row).add(new Put(row))));
1705       fail("Should fail since the put does not contain any cells");
1706     } catch (IllegalArgumentException e) {
1707       assertThat(e.getMessage(), containsString("No columns to insert"));
1708     }
1709
1710     try {
1711       getTable.get().checkAndMutate(CheckAndMutate.newBuilder(row)
1712         .ifNotExists(FAMILY, QUALIFIER)
1713         .build(new RowMutations(row).add(new Put(row)
1714           .addColumn(FAMILY, QUALIFIER, new byte[MAX_KEY_VALUE_SIZE]))));
1715       fail("Should fail since the put exceeds the max key value size");
1716     } catch (IllegalArgumentException e) {
1717       assertThat(e.getMessage(), containsString("KeyValue size too large"));
1718     }
1719   }
1720 }