hbase-server/src/test/java/org/apache/hadoop/hbase/TestClientOperationTimeout.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;
  19
  20 import java.io.IOException;
  21 import java.net.SocketTimeoutException;
  22 import org.apache.hadoop.conf.Configuration;
  23 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
  24 import org.apache.hadoop.hbase.client.Connection;
  25 import org.apache.hadoop.hbase.client.ConnectionFactory;
  26 import org.apache.hadoop.hbase.client.Get;
  27 import org.apache.hadoop.hbase.client.Put;
  28 import org.apache.hadoop.hbase.client.ResultScanner;
  29 import org.apache.hadoop.hbase.client.RetriesExhaustedException;
  30 import org.apache.hadoop.hbase.client.Scan;
  31 import org.apache.hadoop.hbase.client.Table;
  32 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
  33 import org.apache.hadoop.hbase.regionserver.HRegionServer;
  34 import org.apache.hadoop.hbase.regionserver.RSRpcServices;
  35 import org.apache.hadoop.hbase.testclassification.ClientTests;
  36 import org.apache.hadoop.hbase.testclassification.MediumTests;
  37 import org.apache.hadoop.hbase.util.Bytes;
  38 import org.junit.AfterClass;
  39 import org.junit.Before;
  40 import org.junit.BeforeClass;
  41 import org.junit.ClassRule;
  42 import org.junit.Test;
  43 import org.junit.experimental.categories.Category;
  44
  45 import org.apache.hbase.thirdparty.com.google.common.io.Closeables;
  46 import org.apache.hbase.thirdparty.com.google.protobuf.RpcController;
  47 import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException;
  48
  49 import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
  50
  51 /**
  52  * These tests verify that the RPC timeouts ('hbase.client.operation.timeout' and
  53  * 'hbase.client.scanner.timeout.period') work correctly using a modified Region Server which
  54  * injects delays to get, scan and mutate operations.
  55  * <p/>
  56  * When 'hbase.client.operation.timeout' is set and client operation is not completed in time the
  57  * client will retry the operation 'hbase.client.retries.number' times. After that
  58  * {@link SocketTimeoutException} will be thrown.
  59  * <p/>
  60  * Using 'hbase.client.scanner.timeout.period' configuration property similar behavior can be
  61  * specified for scan related operations such as openScanner(), next(). If that times out
  62  * {@link RetriesExhaustedException} will be thrown.
  63  */
  64 @Category({ ClientTests.class, MediumTests.class })
  65 public class TestClientOperationTimeout {
  66
  67   @ClassRule
  68   public static final HBaseClassTestRule CLASS_RULE =
  69     HBaseClassTestRule.forClass(TestClientOperationTimeout.class);
  70
  71   private static final HBaseTestingUtility UTIL = new HBaseTestingUtility();
  72
  73   // Activate the delays after table creation to test get/scan/put
  74   private static int DELAY_GET;
  75   private static int DELAY_SCAN;
  76   private static int DELAY_MUTATE;
  77
  78   private static final TableName TABLE_NAME = TableName.valueOf("Timeout");
  79   private static final byte[] FAMILY = Bytes.toBytes("family");
  80   private static final byte[] ROW = Bytes.toBytes("row");
  81   private static final byte[] QUALIFIER = Bytes.toBytes("qualifier");
  82   private static final byte[] VALUE = Bytes.toBytes("value");
  83
  84   private static Connection CONN;
  85   private static Table TABLE;
  86
  87   @BeforeClass
  88   public static void setUpClass() throws Exception {
  89     // Set RegionServer class and use default values for other options.
  90     StartMiniClusterOption option =
  91       StartMiniClusterOption.builder().rsClass(DelayedRegionServer.class).build();
  92     UTIL.startMiniCluster(option);
  93     UTIL.getAdmin().createTable(TableDescriptorBuilder.newBuilder(TABLE_NAME)
  94       .setColumnFamily(ColumnFamilyDescriptorBuilder.of(FAMILY)).build());
  95
  96     Configuration conf = new Configuration(UTIL.getConfiguration());
  97     conf.setLong(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, 500);
  98     conf.setLong(HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT, 500);
  99     conf.setLong(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, 500);
 100     conf.setLong(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 1);
 101     CONN = ConnectionFactory.createConnection(conf);
 102     TABLE = CONN.getTable(TABLE_NAME);
 103   }
 104
 105   @Before
 106   public void setUp() throws Exception {
 107     DELAY_GET = 0;
 108     DELAY_SCAN = 0;
 109     DELAY_MUTATE = 0;
 110   }
 111
 112   @AfterClass
 113   public static void tearDown() throws Exception {
 114     Closeables.close(TABLE, true);
 115     Closeables.close(CONN, true);
 116     UTIL.shutdownMiniCluster();
 117   }
 118
 119   /**
 120    * Tests that a get on a table throws {@link SocketTimeoutException} when the operation takes
 121    * longer than 'hbase.client.operation.timeout'.
 122    */
 123   @Test(expected = RetriesExhaustedException.class)
 124   public void testGetTimeout() throws Exception {
 125     DELAY_GET = 600;
 126     TABLE.get(new Get(ROW));
 127   }
 128
 129   /**
 130    * Tests that a put on a table throws {@link SocketTimeoutException} when the operation takes
 131    * longer than 'hbase.client.operation.timeout'.
 132    */
 133   @Test(expected = RetriesExhaustedException.class)
 134   public void testPutTimeout() throws Exception {
 135     DELAY_MUTATE = 600;
 136
 137     Put put = new Put(ROW);
 138     put.addColumn(FAMILY, QUALIFIER, VALUE);
 139     TABLE.put(put);
 140   }
 141
 142   /**
 143    * Tests that scan on a table throws {@link RetriesExhaustedException} when the operation takes
 144    * longer than 'hbase.client.scanner.timeout.period'.
 145    */
 146   @Test(expected = RetriesExhaustedException.class)
 147   public void testScanTimeout() throws Exception {
 148     DELAY_SCAN = 600;
 149     ResultScanner scanner = TABLE.getScanner(new Scan());
 150     scanner.next();
 151   }
 152
 153   private static class DelayedRegionServer extends MiniHBaseCluster.MiniHBaseClusterRegionServer {
 154     public DelayedRegionServer(Configuration conf) throws IOException, InterruptedException {
 155       super(conf);
 156     }
 157
 158     @Override
 159     protected RSRpcServices createRpcServices() throws IOException {
 160       return new DelayedRSRpcServices(this);
 161     }
 162   }
 163
 164   /**
 165    * This {@link RSRpcServices} class injects delay for Rpc calls and after executes super methods.
 166    */
 167   public static class DelayedRSRpcServices extends RSRpcServices {
 168     DelayedRSRpcServices(HRegionServer rs) throws IOException {
 169       super(rs);
 170     }
 171
 172     @Override
 173     public ClientProtos.GetResponse get(RpcController controller, ClientProtos.GetRequest request)
 174         throws ServiceException {
 175       try {
 176         Thread.sleep(DELAY_GET);
 177       } catch (InterruptedException e) {
 178         LOG.error("Sleep interrupted during get operation", e);
 179       }
 180       return super.get(controller, request);
 181     }
 182
 183     @Override
 184     public ClientProtos.MutateResponse mutate(RpcController rpcc,
 185         ClientProtos.MutateRequest request) throws ServiceException {
 186       try {
 187         Thread.sleep(DELAY_MUTATE);
 188       } catch (InterruptedException e) {
 189         LOG.error("Sleep interrupted during mutate operation", e);
 190       }
 191       return super.mutate(rpcc, request);
 192     }
 193
 194     @Override
 195     public ClientProtos.ScanResponse scan(RpcController controller,
 196         ClientProtos.ScanRequest request) throws ServiceException {
 197       try {
 198         Thread.sleep(DELAY_SCAN);
 199       } catch (InterruptedException e) {
 200         LOG.error("Sleep interrupted during scan operation", e);
 201       }
 202       return super.scan(controller, request);
 203     }
 204   }
 205 }