hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableScanException.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.instanceOf;
  21 import static org.hamcrest.MatcherAssert.assertThat;
  22 import static org.junit.Assert.assertArrayEquals;
  23 import static org.junit.Assert.assertTrue;
  24 import static org.junit.Assert.fail;
  25
  26 import java.io.IOException;
  27 import java.util.List;
  28 import java.util.Optional;
  29 import java.util.concurrent.atomic.AtomicInteger;
  30 import org.apache.hadoop.hbase.DoNotRetryIOException;
  31 import org.apache.hadoop.hbase.HBaseClassTestRule;
  32 import org.apache.hadoop.hbase.HBaseTestingUtil;
  33 import org.apache.hadoop.hbase.TableName;
  34 import org.apache.hadoop.hbase.coprocessor.ObserverContext;
  35 import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
  36 import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
  37 import org.apache.hadoop.hbase.coprocessor.RegionObserver;
  38 import org.apache.hadoop.hbase.exceptions.ScannerResetException;
  39 import org.apache.hadoop.hbase.regionserver.InternalScanner;
  40 import org.apache.hadoop.hbase.testclassification.ClientTests;
  41 import org.apache.hadoop.hbase.testclassification.MediumTests;
  42 import org.apache.hadoop.hbase.util.Bytes;
  43 import org.junit.AfterClass;
  44 import org.junit.Before;
  45 import org.junit.BeforeClass;
  46 import org.junit.ClassRule;
  47 import org.junit.Test;
  48 import org.junit.experimental.categories.Category;
  49
  50 import org.apache.hbase.thirdparty.com.google.common.io.Closeables;
  51
  52 @Category({ MediumTests.class, ClientTests.class })
  53 public class TestAsyncTableScanException {
  54
  55   @ClassRule
  56   public static final HBaseClassTestRule CLASS_RULE =
  57     HBaseClassTestRule.forClass(TestAsyncTableScanException.class);
  58
  59   private static final HBaseTestingUtil UTIL = new HBaseTestingUtil();
  60
  61   private static TableName TABLE_NAME = TableName.valueOf("scan");
  62
  63   private static byte[] FAMILY = Bytes.toBytes("family");
  64
  65   private static byte[] QUAL = Bytes.toBytes("qual");
  66
  67   private static AsyncConnection CONN;
  68
  69   private static AtomicInteger REQ_COUNT = new AtomicInteger();
  70
  71   private static volatile int ERROR_AT;
  72
  73   private static volatile boolean ERROR;
  74
  75   private static volatile boolean DO_NOT_RETRY;
  76
  77   private static final int ROW_COUNT = 100;
  78
  79   public static final class ErrorCP implements RegionObserver, RegionCoprocessor {
  80
  81     @Override
  82     public Optional<RegionObserver> getRegionObserver() {
  83       return Optional.of(this);
  84     }
  85
  86     @Override
  87     public boolean postScannerNext(ObserverContext<RegionCoprocessorEnvironment> c,
  88         InternalScanner s, List<Result> result, int limit, boolean hasNext) throws IOException {
  89       REQ_COUNT.incrementAndGet();
  90       if ((ERROR_AT == REQ_COUNT.get()) || ERROR) {
  91         if (DO_NOT_RETRY) {
  92           throw new DoNotRetryIOException("Injected exception");
  93         } else {
  94           throw new IOException("Injected exception");
  95         }
  96       }
  97       return RegionObserver.super.postScannerNext(c, s, result, limit, hasNext);
  98     }
  99
 100   }
 101
 102   @BeforeClass
 103   public static void setUp() throws Exception {
 104     UTIL.startMiniCluster(1);
 105     UTIL.getAdmin()
 106       .createTable(TableDescriptorBuilder.newBuilder(TABLE_NAME)
 107         .setColumnFamily(ColumnFamilyDescriptorBuilder.of(FAMILY))
 108         .setCoprocessor(ErrorCP.class.getName()).build());
 109     try (Table table = UTIL.getConnection().getTable(TABLE_NAME)) {
 110       for (int i = 0; i < ROW_COUNT; i++) {
 111         table.put(new Put(Bytes.toBytes(i)).addColumn(FAMILY, QUAL, Bytes.toBytes(i)));
 112       }
 113     }
 114     CONN = ConnectionFactory.createAsyncConnection(UTIL.getConfiguration()).get();
 115   }
 116
 117   @AfterClass
 118   public static void tearDown() throws Exception {
 119     Closeables.close(CONN, true);
 120     UTIL.shutdownMiniCluster();
 121   }
 122
 123   @Before
 124   public void setUpBeforeTest() {
 125     REQ_COUNT.set(0);
 126     ERROR_AT = 0;
 127     ERROR = false;
 128     DO_NOT_RETRY = false;
 129   }
 130
 131   @Test(expected = DoNotRetryIOException.class)
 132   public void testDoNotRetryIOException() throws IOException {
 133     ERROR_AT = 1;
 134     DO_NOT_RETRY = true;
 135     try (ResultScanner scanner = CONN.getTable(TABLE_NAME).getScanner(FAMILY)) {
 136       scanner.next();
 137     }
 138   }
 139
 140   @Test
 141   public void testIOException() throws IOException {
 142     ERROR = true;
 143     try (ResultScanner scanner =
 144       CONN.getTableBuilder(TABLE_NAME).setMaxAttempts(3).build().getScanner(FAMILY)) {
 145       scanner.next();
 146       fail();
 147     } catch (RetriesExhaustedException e) {
 148       // expected
 149       assertThat(e.getCause(), instanceOf(ScannerResetException.class));
 150     }
 151     assertTrue(REQ_COUNT.get() >= 3);
 152   }
 153
 154   private void count() throws IOException {
 155     try (ResultScanner scanner = CONN.getTable(TABLE_NAME).getScanner(new Scan().setCaching(1))) {
 156       for (int i = 0; i < ROW_COUNT; i++) {
 157         Result result = scanner.next();
 158         assertArrayEquals(Bytes.toBytes(i), result.getRow());
 159         assertArrayEquals(Bytes.toBytes(i), result.getValue(FAMILY, QUAL));
 160       }
 161     }
 162   }
 163
 164   @Test
 165   public void testRecoveryFromScannerResetWhileOpening() throws IOException {
 166     ERROR_AT = 1;
 167     count();
 168     // we should at least request 1 time otherwise the error will not be triggered, and then we
 169     // need at least one more request to get the remaining results.
 170     assertTrue(REQ_COUNT.get() >= 2);
 171   }
 172
 173   @Test
 174   public void testRecoveryFromScannerResetInTheMiddle() throws IOException {
 175     ERROR_AT = 2;
 176     count();
 177     // we should at least request 2 times otherwise the error will not be triggered, and then we
 178     // need at least one more request to get the remaining results.
 179     assertTrue(REQ_COUNT.get() >= 3);
 180   }
 181 }