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.assertTrue; 023 024import java.io.IOException; 025import java.util.Optional; 026import java.util.concurrent.ExecutionException; 027import java.util.concurrent.TimeUnit; 028import java.util.concurrent.atomic.AtomicInteger; 029import org.apache.hadoop.hbase.HBaseClassTestRule; 030import org.apache.hadoop.hbase.HBaseTestingUtility; 031import org.apache.hadoop.hbase.TableName; 032import org.apache.hadoop.hbase.coprocessor.ObserverContext; 033import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor; 034import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; 035import org.apache.hadoop.hbase.coprocessor.RegionObserver; 036import org.apache.hadoop.hbase.testclassification.ClientTests; 037import org.apache.hadoop.hbase.testclassification.MediumTests; 038import org.apache.hadoop.hbase.util.Bytes; 039import org.apache.hadoop.hbase.util.Threads; 040import org.junit.AfterClass; 041import org.junit.Before; 042import org.junit.BeforeClass; 043import org.junit.ClassRule; 044import org.junit.Rule; 045import org.junit.Test; 046import org.junit.experimental.categories.Category; 047import org.junit.rules.TestName; 048 049import org.apache.hbase.thirdparty.com.google.common.io.Closeables; 050 051@Category({ MediumTests.class, ClientTests.class }) 052public class TestAsyncTableNoncedRetry { 053 054 @ClassRule 055 public static final HBaseClassTestRule CLASS_RULE = 056 HBaseClassTestRule.forClass(TestAsyncTableNoncedRetry.class); 057 058 private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility(); 059 060 private static TableName TABLE_NAME = TableName.valueOf("async"); 061 062 private static byte[] FAMILY = Bytes.toBytes("cf"); 063 064 private static byte[] QUALIFIER = Bytes.toBytes("cq"); 065 066 private static byte[] VALUE = Bytes.toBytes("value"); 067 068 private static AsyncConnection ASYNC_CONN; 069 070 @Rule 071 public TestName testName = new TestName(); 072 073 private byte[] row; 074 075 private static AtomicInteger CALLED = new AtomicInteger(); 076 077 private static long SLEEP_TIME = 2000; 078 079 public static final class SleepOnceCP implements RegionObserver, RegionCoprocessor { 080 081 @Override 082 public Optional<RegionObserver> getRegionObserver() { 083 return Optional.of(this); 084 } 085 086 @Override 087 public Result postAppend(ObserverContext<RegionCoprocessorEnvironment> c, Append append, 088 Result result) throws IOException { 089 if (CALLED.getAndIncrement() == 0) { 090 Threads.sleepWithoutInterrupt(SLEEP_TIME); 091 } 092 return RegionObserver.super.postAppend(c, append, result); 093 } 094 095 @Override 096 public Result postIncrement(ObserverContext<RegionCoprocessorEnvironment> c, 097 Increment increment, Result result) throws IOException { 098 if (CALLED.getAndIncrement() == 0) { 099 Threads.sleepWithoutInterrupt(SLEEP_TIME); 100 } 101 return RegionObserver.super.postIncrement(c, increment, result); 102 } 103 } 104 105 @BeforeClass 106 public static void setUpBeforeClass() throws Exception { 107 TEST_UTIL.startMiniCluster(1); 108 TEST_UTIL.getAdmin() 109 .createTable(TableDescriptorBuilder.newBuilder(TABLE_NAME) 110 .setColumnFamily(ColumnFamilyDescriptorBuilder.of(FAMILY)) 111 .setCoprocessor(SleepOnceCP.class.getName()).build()); 112 TEST_UTIL.waitTableAvailable(TABLE_NAME); 113 ASYNC_CONN = ConnectionFactory.createAsyncConnection(TEST_UTIL.getConfiguration()).get(); 114 } 115 116 @AfterClass 117 public static void tearDownAfterClass() throws Exception { 118 Closeables.close(ASYNC_CONN, true); 119 TEST_UTIL.shutdownMiniCluster(); 120 } 121 122 @Before 123 public void setUp() throws IOException, InterruptedException { 124 row = Bytes.toBytes(testName.getMethodName().replaceAll("[^0-9A-Za-z]", "_")); 125 CALLED.set(0); 126 } 127 128 @Test 129 public void testAppend() throws InterruptedException, ExecutionException { 130 assertEquals(0, CALLED.get()); 131 AsyncTable<?> table = ASYNC_CONN.getTableBuilder(TABLE_NAME) 132 .setRpcTimeout(SLEEP_TIME / 2, TimeUnit.MILLISECONDS).build(); 133 Result result = table.append(new Append(row).addColumn(FAMILY, QUALIFIER, VALUE)).get(); 134 // make sure we called twice and the result is still correct 135 assertEquals(2, CALLED.get()); 136 assertArrayEquals(VALUE, result.getValue(FAMILY, QUALIFIER)); 137 } 138 139 @Test 140 public void testAppendWhenReturnResultsEqualsFalse() throws InterruptedException, 141 ExecutionException { 142 assertEquals(0, CALLED.get()); 143 AsyncTable<?> table = ASYNC_CONN.getTableBuilder(TABLE_NAME) 144 .setRpcTimeout(SLEEP_TIME / 2, TimeUnit.MILLISECONDS).build(); 145 Result result = table.append(new Append(row).addColumn(FAMILY, QUALIFIER, VALUE) 146 .setReturnResults(false)).get(); 147 // make sure we called twice and the result is still correct 148 assertEquals(2, CALLED.get()); 149 assertTrue(result.isEmpty()); 150 } 151 152 @Test 153 public void testIncrement() throws InterruptedException, ExecutionException { 154 assertEquals(0, CALLED.get()); 155 AsyncTable<?> table = ASYNC_CONN.getTableBuilder(TABLE_NAME) 156 .setRpcTimeout(SLEEP_TIME / 2, TimeUnit.MILLISECONDS).build(); 157 assertEquals(1L, table.incrementColumnValue(row, FAMILY, QUALIFIER, 1L).get().longValue()); 158 // make sure we called twice and the result is still correct 159 assertEquals(2, CALLED.get()); 160 } 161 162 @Test 163 public void testIncrementWhenReturnResultsEqualsFalse() throws InterruptedException, 164 ExecutionException { 165 assertEquals(0, CALLED.get()); 166 AsyncTable<?> table = ASYNC_CONN.getTableBuilder(TABLE_NAME) 167 .setRpcTimeout(SLEEP_TIME / 2, TimeUnit.MILLISECONDS).build(); 168 Result result = table.increment(new Increment(row).addColumn(FAMILY, QUALIFIER, 1L) 169 .setReturnResults(false)).get(); 170 // make sure we called twice and the result is still correct 171 assertEquals(2, CALLED.get()); 172 assertTrue(result.isEmpty()); 173 } 174}