Source code

001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.client;
019
020import static org.junit.Assert.assertArrayEquals;
021import static org.junit.Assert.assertEquals;
022import static org.junit.Assert.assertFalse;
023import static org.junit.Assert.assertNull;
024import static org.junit.Assert.assertTrue;
025
026import java.io.IOException;
027import java.io.UncheckedIOException;
028import java.util.Arrays;
029import java.util.List;
030import java.util.concurrent.ArrayBlockingQueue;
031import java.util.concurrent.BlockingQueue;
032import java.util.concurrent.CountDownLatch;
033import java.util.concurrent.ExecutionException;
034import java.util.concurrent.ForkJoinPool;
035import java.util.concurrent.atomic.AtomicInteger;
036import java.util.concurrent.atomic.AtomicLong;
037import java.util.function.Supplier;
038import java.util.stream.IntStream;
039import org.apache.commons.io.IOUtils;
040import org.apache.hadoop.hbase.HBaseClassTestRule;
041import org.apache.hadoop.hbase.HBaseTestingUtility;
042import org.apache.hadoop.hbase.TableName;
043import org.apache.hadoop.hbase.io.TimeRange;
044import org.apache.hadoop.hbase.testclassification.ClientTests;
045import org.apache.hadoop.hbase.testclassification.MediumTests;
046import org.apache.hadoop.hbase.util.Bytes;
047import org.apache.hadoop.hbase.util.Pair;
048import org.junit.AfterClass;
049import org.junit.Before;
050import org.junit.BeforeClass;
051import org.junit.ClassRule;
052import org.junit.Rule;
053import org.junit.Test;
054import org.junit.experimental.categories.Category;
055import org.junit.rules.TestName;
056import org.junit.runner.RunWith;
057import org.junit.runners.Parameterized;
058import org.junit.runners.Parameterized.Parameter;
059import org.junit.runners.Parameterized.Parameters;
060
061@RunWith(Parameterized.class)
062@Category({ MediumTests.class, ClientTests.class })
063public class TestAsyncTable {
064
065  @ClassRule
066  public static final HBaseClassTestRule CLASS_RULE =
067      HBaseClassTestRule.forClass(TestAsyncTable.class);
068
069  private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
070
071  private static TableName TABLE_NAME = TableName.valueOf("async");
072
073  private static byte[] FAMILY = Bytes.toBytes("cf");
074
075  private static byte[] QUALIFIER = Bytes.toBytes("cq");
076
077  private static byte[] VALUE = Bytes.toBytes("value");
078
079  private static AsyncConnection ASYNC_CONN;
080
081  @Rule
082  public TestName testName = new TestName();
083
084  private byte[] row;
085
086  @Parameter
087  public Supplier<AsyncTable<?>> getTable;
088
089  private static AsyncTable<?> getRawTable() {
090    return ASYNC_CONN.getTable(TABLE_NAME);
091  }
092
093  private static AsyncTable<?> getTable() {
094    return ASYNC_CONN.getTable(TABLE_NAME, ForkJoinPool.commonPool());
095  }
096
097  @Parameters
098  public static List<Object[]> params() {
099    return Arrays.asList(new Supplier<?>[] { TestAsyncTable::getRawTable },
100      new Supplier<?>[] { TestAsyncTable::getTable });
101  }
102
103  @BeforeClass
104  public static void setUpBeforeClass() throws Exception {
105    TEST_UTIL.startMiniCluster(1);
106    TEST_UTIL.createTable(TABLE_NAME, FAMILY);
107    TEST_UTIL.waitTableAvailable(TABLE_NAME);
108    ASYNC_CONN = ConnectionFactory.createAsyncConnection(TEST_UTIL.getConfiguration()).get();
109  }
110
111  @AfterClass
112  public static void tearDownAfterClass() throws Exception {
113    IOUtils.closeQuietly(ASYNC_CONN);
114    TEST_UTIL.shutdownMiniCluster();
115  }
116
117  @Before
118  public void setUp() throws IOException, InterruptedException {
119    row = Bytes.toBytes(testName.getMethodName().replaceAll("[^0-9A-Za-z]", "_"));
120  }
121
122  @Test
123  public void testSimple() throws Exception {
124    AsyncTable<?> table = getTable.get();
125    table.put(new Put(row).addColumn(FAMILY, QUALIFIER, VALUE)).get();
126    assertTrue(table.exists(new Get(row).addColumn(FAMILY, QUALIFIER)).get());
127    Result result = table.get(new Get(row).addColumn(FAMILY, QUALIFIER)).get();
128    assertArrayEquals(VALUE, result.getValue(FAMILY, QUALIFIER));
129    table.delete(new Delete(row)).get();
130    result = table.get(new Get(row).addColumn(FAMILY, QUALIFIER)).get();
131    assertTrue(result.isEmpty());
132    assertFalse(table.exists(new Get(row).addColumn(FAMILY, QUALIFIER)).get());
133  }
134
135  private byte[] concat(byte[] base, int index) {
136    return Bytes.toBytes(Bytes.toString(base) + "-" + index);
137  }
138
139  @Test
140  public void testSimpleMultiple() throws Exception {
141    AsyncTable<?> table = getTable.get();
142    int count = 100;
143    CountDownLatch putLatch = new CountDownLatch(count);
144    IntStream.range(0, count).forEach(
145      i -> table.put(new Put(concat(row, i)).addColumn(FAMILY, QUALIFIER, concat(VALUE, i)))
146          .thenAccept(x -> putLatch.countDown()));
147    putLatch.await();
148    BlockingQueue<Boolean> existsResp = new ArrayBlockingQueue<>(count);
149    IntStream.range(0, count)
150        .forEach(i -> table.exists(new Get(concat(row, i)).addColumn(FAMILY, QUALIFIER))
151            .thenAccept(x -> existsResp.add(x)));
152    for (int i = 0; i < count; i++) {
153      assertTrue(existsResp.take());
154    }
155    BlockingQueue<Pair<Integer, Result>> getResp = new ArrayBlockingQueue<>(count);
156    IntStream.range(0, count)
157        .forEach(i -> table.get(new Get(concat(row, i)).addColumn(FAMILY, QUALIFIER))
158            .thenAccept(x -> getResp.add(Pair.newPair(i, x))));
159    for (int i = 0; i < count; i++) {
160      Pair<Integer, Result> pair = getResp.take();
161      assertArrayEquals(concat(VALUE, pair.getFirst()),
162        pair.getSecond().getValue(FAMILY, QUALIFIER));
163    }
164    CountDownLatch deleteLatch = new CountDownLatch(count);
165    IntStream.range(0, count).forEach(
166      i -> table.delete(new Delete(concat(row, i))).thenAccept(x -> deleteLatch.countDown()));
167    deleteLatch.await();
168    IntStream.range(0, count)
169        .forEach(i -> table.exists(new Get(concat(row, i)).addColumn(FAMILY, QUALIFIER))
170            .thenAccept(x -> existsResp.add(x)));
171    for (int i = 0; i < count; i++) {
172      assertFalse(existsResp.take());
173    }
174    IntStream.range(0, count)
175        .forEach(i -> table.get(new Get(concat(row, i)).addColumn(FAMILY, QUALIFIER))
176            .thenAccept(x -> getResp.add(Pair.newPair(i, x))));
177    for (int i = 0; i < count; i++) {
178      Pair<Integer, Result> pair = getResp.take();
179      assertTrue(pair.getSecond().isEmpty());
180    }
181  }
182
183  @Test
184  public void testIncrement() throws InterruptedException, ExecutionException {
185    AsyncTable<?> table = getTable.get();
186    int count = 100;
187    CountDownLatch latch = new CountDownLatch(count);
188    AtomicLong sum = new AtomicLong(0L);
189    IntStream.range(0, count)
190        .forEach(i -> table.incrementColumnValue(row, FAMILY, QUALIFIER, 1).thenAccept(x -> {
191          sum.addAndGet(x);
192          latch.countDown();
193        }));
194    latch.await();
195    assertEquals(count, Bytes.toLong(
196      table.get(new Get(row).addColumn(FAMILY, QUALIFIER)).get().getValue(FAMILY, QUALIFIER)));
197    assertEquals((1 + count) * count / 2, sum.get());
198  }
199
200  @Test
201  public void testAppend() throws InterruptedException, ExecutionException {
202    AsyncTable<?> table = getTable.get();
203    int count = 10;
204    CountDownLatch latch = new CountDownLatch(count);
205    char suffix = ':';
206    AtomicLong suffixCount = new AtomicLong(0L);
207    IntStream.range(0, count).forEachOrdered(
208      i -> table.append(new Append(row).addColumn(FAMILY, QUALIFIER, Bytes.toBytes("" + i + suffix)))
209          .thenAccept(r -> {
210            suffixCount.addAndGet(Bytes.toString(r.getValue(FAMILY, QUALIFIER)).chars()
211                .filter(x -> x == suffix).count());
212            latch.countDown();
213          }));
214    latch.await();
215    assertEquals((1 + count) * count / 2, suffixCount.get());
216    String value = Bytes.toString(
217      table.get(new Get(row).addColumn(FAMILY, QUALIFIER)).get().getValue(FAMILY, QUALIFIER));
218    int[] actual = Arrays.asList(value.split("" + suffix)).stream().mapToInt(Integer::parseInt)
219        .sorted().toArray();
220    assertArrayEquals(IntStream.range(0, count).toArray(), actual);
221  }
222
223  @Test
224  public void testCheckAndPut() throws InterruptedException, ExecutionException {
225    AsyncTable<?> table = getTable.get();
226    AtomicInteger successCount = new AtomicInteger(0);
227    AtomicInteger successIndex = new AtomicInteger(-1);
228    int count = 10;
229    CountDownLatch latch = new CountDownLatch(count);
230    IntStream.range(0, count)
231        .forEach(i -> table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER).ifNotExists()
232            .thenPut(new Put(row).addColumn(FAMILY, QUALIFIER, concat(VALUE, i))).thenAccept(x -> {
233              if (x) {
234                successCount.incrementAndGet();
235                successIndex.set(i);
236              }
237              latch.countDown();
238            }));
239    latch.await();
240    assertEquals(1, successCount.get());
241    String actual = Bytes.toString(table.get(new Get(row)).get().getValue(FAMILY, QUALIFIER));
242    assertTrue(actual.endsWith(Integer.toString(successIndex.get())));
243  }
244
245  @Test
246  public void testCheckAndDelete() throws InterruptedException, ExecutionException {
247    AsyncTable<?> table = getTable.get();
248    int count = 10;
249    CountDownLatch putLatch = new CountDownLatch(count + 1);
250    table.put(new Put(row).addColumn(FAMILY, QUALIFIER, VALUE)).thenRun(() -> putLatch.countDown());
251    IntStream.range(0, count)
252        .forEach(i -> table.put(new Put(row).addColumn(FAMILY, concat(QUALIFIER, i), VALUE))
253            .thenRun(() -> putLatch.countDown()));
254    putLatch.await();
255
256    AtomicInteger successCount = new AtomicInteger(0);
257    AtomicInteger successIndex = new AtomicInteger(-1);
258    CountDownLatch deleteLatch = new CountDownLatch(count);
259    IntStream.range(0, count)
260        .forEach(i -> table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER).ifEquals(VALUE)
261            .thenDelete(
262              new Delete(row).addColumn(FAMILY, QUALIFIER).addColumn(FAMILY, concat(QUALIFIER, i)))
263            .thenAccept(x -> {
264              if (x) {
265                successCount.incrementAndGet();
266                successIndex.set(i);
267              }
268              deleteLatch.countDown();
269            }));
270    deleteLatch.await();
271    assertEquals(1, successCount.get());
272    Result result = table.get(new Get(row)).get();
273    IntStream.range(0, count).forEach(i -> {
274      if (i == successIndex.get()) {
275        assertFalse(result.containsColumn(FAMILY, concat(QUALIFIER, i)));
276      } else {
277        assertArrayEquals(VALUE, result.getValue(FAMILY, concat(QUALIFIER, i)));
278      }
279    });
280  }
281
282  @Test
283  public void testMutateRow() throws InterruptedException, ExecutionException, IOException {
284    AsyncTable<?> table = getTable.get();
285    RowMutations mutation = new RowMutations(row);
286    mutation.add(new Put(row).addColumn(FAMILY, concat(QUALIFIER, 1), VALUE));
287    table.mutateRow(mutation).get();
288    Result result = table.get(new Get(row)).get();
289    assertArrayEquals(VALUE, result.getValue(FAMILY, concat(QUALIFIER, 1)));
290
291    mutation = new RowMutations(row);
292    mutation.add(new Delete(row).addColumn(FAMILY, concat(QUALIFIER, 1)));
293    mutation.add(new Put(row).addColumn(FAMILY, concat(QUALIFIER, 2), VALUE));
294    table.mutateRow(mutation).get();
295    result = table.get(new Get(row)).get();
296    assertNull(result.getValue(FAMILY, concat(QUALIFIER, 1)));
297    assertArrayEquals(VALUE, result.getValue(FAMILY, concat(QUALIFIER, 2)));
298  }
299
300  @Test
301  public void testCheckAndMutate() throws InterruptedException, ExecutionException {
302    AsyncTable<?> table = getTable.get();
303    int count = 10;
304    CountDownLatch putLatch = new CountDownLatch(count + 1);
305    table.put(new Put(row).addColumn(FAMILY, QUALIFIER, VALUE)).thenRun(() -> putLatch.countDown());
306    IntStream.range(0, count)
307        .forEach(i -> table.put(new Put(row).addColumn(FAMILY, concat(QUALIFIER, i), VALUE))
308            .thenRun(() -> putLatch.countDown()));
309    putLatch.await();
310
311    AtomicInteger successCount = new AtomicInteger(0);
312    AtomicInteger successIndex = new AtomicInteger(-1);
313    CountDownLatch mutateLatch = new CountDownLatch(count);
314    IntStream.range(0, count).forEach(i -> {
315      RowMutations mutation = new RowMutations(row);
316      try {
317        mutation.add(new Delete(row).addColumn(FAMILY, QUALIFIER));
318        mutation.add(new Put(row).addColumn(FAMILY, concat(QUALIFIER, i), concat(VALUE, i)));
319      } catch (IOException e) {
320        throw new UncheckedIOException(e);
321      }
322      table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER).ifEquals(VALUE).thenMutate(mutation)
323          .thenAccept(x -> {
324            if (x) {
325              successCount.incrementAndGet();
326              successIndex.set(i);
327            }
328            mutateLatch.countDown();
329          });
330    });
331    mutateLatch.await();
332    assertEquals(1, successCount.get());
333    Result result = table.get(new Get(row)).get();
334    IntStream.range(0, count).forEach(i -> {
335      if (i == successIndex.get()) {
336        assertArrayEquals(concat(VALUE, i), result.getValue(FAMILY, concat(QUALIFIER, i)));
337      } else {
338        assertArrayEquals(VALUE, result.getValue(FAMILY, concat(QUALIFIER, i)));
339      }
340    });
341  }
342
343  @Test
344  public void testCheckAndMutateWithTimeRange() throws Exception {
345    AsyncTable<?> table = getTable.get();
346    final long ts = System.currentTimeMillis() / 2;
347    Put put = new Put(row);
348    put.addColumn(FAMILY, QUALIFIER, ts, VALUE);
349
350    boolean ok = table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER)
351      .ifNotExists()
352      .thenPut(put)
353      .get();
354    assertTrue(ok);
355
356    ok = table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER)
357      .timeRange(TimeRange.at(ts + 10000))
358      .ifEquals(VALUE)
359      .thenPut(put)
360      .get();
361    assertFalse(ok);
362
363    ok = table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER)
364      .timeRange(TimeRange.at(ts))
365      .ifEquals(VALUE)
366      .thenPut(put)
367      .get();
368    assertTrue(ok);
369
370    RowMutations rm = new RowMutations(row)
371      .add((Mutation) put);
372    ok = table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER)
373      .timeRange(TimeRange.at(ts + 10000))
374      .ifEquals(VALUE)
375      .thenMutate(rm)
376      .get();
377    assertFalse(ok);
378
379    ok = table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER)
380      .timeRange(TimeRange.at(ts))
381      .ifEquals(VALUE)
382      .thenMutate(rm)
383      .get();
384    assertTrue(ok);
385
386    Delete delete = new Delete(row)
387      .addColumn(FAMILY, QUALIFIER);
388
389    ok = table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER)
390      .timeRange(TimeRange.at(ts + 10000))
391      .ifEquals(VALUE)
392      .thenDelete(delete)
393      .get();
394    assertFalse(ok);
395
396    ok = table.checkAndMutate(row, FAMILY).qualifier(QUALIFIER)
397      .timeRange(TimeRange.at(ts))
398      .ifEquals(VALUE)
399      .thenDelete(delete)
400      .get();
401    assertTrue(ok);
402  }
403}