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;
019
020import java.io.IOException;
021
022import java.net.SocketTimeoutException;
023import org.apache.hadoop.conf.Configuration;
024import org.apache.hadoop.hbase.client.Get;
025import org.apache.hadoop.hbase.client.Put;
026import org.apache.hadoop.hbase.client.ResultScanner;
027import org.apache.hadoop.hbase.client.RetriesExhaustedException;
028import org.apache.hadoop.hbase.client.Scan;
029import org.apache.hadoop.hbase.client.Table;
030import org.apache.hadoop.hbase.regionserver.HRegionServer;
031import org.apache.hadoop.hbase.regionserver.RSRpcServices;
032import org.apache.hadoop.hbase.testclassification.ClientTests;
033import org.apache.hadoop.hbase.testclassification.MediumTests;
034import org.apache.hadoop.hbase.util.Bytes;
035import org.apache.hbase.thirdparty.com.google.protobuf.RpcController;
036import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException;
037import org.junit.AfterClass;
038import org.junit.Before;
039import org.junit.BeforeClass;
040import org.junit.ClassRule;
041import org.junit.Rule;
042import org.junit.Test;
043import org.junit.experimental.categories.Category;
044import org.junit.rules.TestName;
045
046import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
047
048/**
049 * These tests verify that the RPC timeouts ('hbase.client.operation.timeout' and
050 * 'hbase.client.scanner.timeout.period') work correctly using a modified Region Server which
051 * injects delays to get, scan and mutate operations.
052 *
053 * When 'hbase.client.operation.timeout' is set and client operation is not completed in time the
054 * client will retry the operation 'hbase.client.retries.number' times. After that
055 * {@link SocketTimeoutException} will be thrown.
056 *
057 * Using 'hbase.client.scanner.timeout.period' configuration property similar behavior can be
058 * specified for scan related operations such as openScanner(), next(). If that times out
059 * {@link RetriesExhaustedException} will be thrown.
060 */
061@Category({ClientTests.class, MediumTests.class})
062public class TestClientOperationTimeout {
063
064  @ClassRule
065  public static final HBaseClassTestRule CLASS_RULE =
066      HBaseClassTestRule.forClass(TestClientOperationTimeout.class);
067
068  private static final HBaseTestingUtility TESTING_UTIL = new HBaseTestingUtility();
069
070  // Activate the delays after table creation to test get/scan/put
071  private static int DELAY_GET;
072  private static int DELAY_SCAN;
073  private static int DELAY_MUTATE;
074
075  private final byte[] FAMILY = Bytes.toBytes("family");
076  private final byte[] ROW = Bytes.toBytes("row");
077  private final byte[] QUALIFIER = Bytes.toBytes("qualifier");
078  private final byte[] VALUE = Bytes.toBytes("value");
079
080  @Rule
081  public TestName name = new TestName();
082  private Table table;
083
084  @BeforeClass
085  public static void setUpClass() throws Exception {
086    TESTING_UTIL.getConfiguration().setLong(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, 500);
087    TESTING_UTIL.getConfiguration().setLong(HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT, 500);
088    TESTING_UTIL.getConfiguration().setLong(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, 500);
089    TESTING_UTIL.getConfiguration().setLong(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 1);
090
091    TESTING_UTIL.startMiniCluster(1, 1, null, null, DelayedRegionServer.class);
092  }
093
094  @Before
095  public void setUp() throws Exception {
096    DELAY_GET = 0;
097    DELAY_SCAN = 0;
098    DELAY_MUTATE = 0;
099
100    table = TESTING_UTIL.createTable(TableName.valueOf(name.getMethodName()), FAMILY);
101    Put put = new Put(ROW);
102    put.addColumn(FAMILY, QUALIFIER, VALUE);
103    table.put(put);
104  }
105
106  @AfterClass
107  public static void tearDown() throws Exception {
108    TESTING_UTIL.shutdownMiniCluster();
109  }
110
111  /**
112   * Tests that a get on a table throws {@link SocketTimeoutException} when the operation takes
113   * longer than 'hbase.client.operation.timeout'.
114   */
115  @Test(expected = SocketTimeoutException.class)
116  public void testGetTimeout() throws Exception {
117    DELAY_GET = 600;
118    table.get(new Get(ROW));
119  }
120
121  /**
122   * Tests that a put on a table throws {@link SocketTimeoutException} when the operation takes
123   * longer than 'hbase.client.operation.timeout'.
124   */
125  @Test(expected = SocketTimeoutException.class)
126  public void testPutTimeout() throws Exception {
127    DELAY_MUTATE = 600;
128
129    Put put = new Put(ROW);
130    put.addColumn(FAMILY, QUALIFIER, VALUE);
131    table.put(put);
132  }
133
134  /**
135   * Tests that scan on a table throws {@link RetriesExhaustedException} when the operation takes
136   * longer than 'hbase.client.scanner.timeout.period'.
137   */
138  @Test(expected = RetriesExhaustedException.class)
139  public void testScanTimeout() throws Exception {
140    DELAY_SCAN = 600;
141    ResultScanner scanner = table.getScanner(new Scan());
142    scanner.next();
143  }
144
145  private static class DelayedRegionServer extends MiniHBaseCluster.MiniHBaseClusterRegionServer {
146    public DelayedRegionServer(Configuration conf) throws IOException, InterruptedException {
147      super(conf);
148    }
149
150    @Override
151    protected RSRpcServices createRpcServices() throws IOException {
152      return new DelayedRSRpcServices(this);
153    }
154  }
155
156  /**
157   * This {@link RSRpcServices} class injects delay for Rpc calls and after executes super methods.
158   */
159  public static class DelayedRSRpcServices extends RSRpcServices {
160    DelayedRSRpcServices(HRegionServer rs) throws IOException {
161      super(rs);
162    }
163
164    @Override
165    public ClientProtos.GetResponse get(RpcController controller, ClientProtos.GetRequest request)
166        throws ServiceException {
167      try {
168        Thread.sleep(DELAY_GET);
169      } catch (InterruptedException e) {
170        LOG.error("Sleep interrupted during get operation", e);
171      }
172      return super.get(controller, request);
173    }
174
175    @Override
176    public ClientProtos.MutateResponse mutate(RpcController rpcc,
177        ClientProtos.MutateRequest request) throws ServiceException {
178      try {
179        Thread.sleep(DELAY_MUTATE);
180      } catch (InterruptedException e) {
181        LOG.error("Sleep interrupted during mutate operation", e);
182      }
183      return super.mutate(rpcc, request);
184    }
185
186    @Override
187    public ClientProtos.ScanResponse scan(RpcController controller,
188        ClientProtos.ScanRequest request) throws ServiceException {
189      try {
190        Thread.sleep(DELAY_SCAN);
191      } catch (InterruptedException e) {
192        LOG.error("Sleep interrupted during scan operation", e);
193      }
194      return super.scan(controller, request);
195    }
196  }
197}