View Javadoc

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 com.google.protobuf.RpcController;
21  import com.google.protobuf.ServiceException;
22  import org.apache.hadoop.conf.Configuration;
23  import org.apache.hadoop.hbase.*;
24  
25  import org.apache.hadoop.hbase.exceptions.ClientExceptionsUtil;
26  import org.apache.hadoop.hbase.exceptions.RegionOpeningException;
27  import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
28  import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.GetResponse;
29  import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
30  import org.apache.hadoop.hbase.quotas.ThrottlingException;
31  import org.apache.hadoop.hbase.regionserver.HRegionServer;
32  import org.apache.hadoop.hbase.regionserver.RSRpcServices;
33  import org.apache.hadoop.hbase.regionserver.Region;
34  import org.apache.hadoop.hbase.testclassification.ClientTests;
35  import org.apache.hadoop.hbase.testclassification.MediumTests;
36  import org.apache.hadoop.hbase.util.Bytes;
37  import org.junit.AfterClass;
38  import org.junit.BeforeClass;
39  import org.junit.Test;
40  import org.junit.experimental.categories.Category;
41  
42  import java.io.IOException;
43  import java.util.ArrayList;
44  import java.util.List;
45  
46  import static junit.framework.Assert.assertEquals;
47  import static org.junit.Assert.assertNotNull;
48  import static org.junit.Assert.assertNull;
49  import static org.junit.Assert.assertTrue;
50  import static org.junit.Assert.fail;
51  
52  @Category({MediumTests.class, ClientTests.class})
53  public class TestMetaCache {
54    private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
55    private static final TableName TABLE_NAME = TableName.valueOf("test_table");
56    private static final byte[] FAMILY = Bytes.toBytes("fam1");
57    private static final byte[] QUALIFIER = Bytes.toBytes("qual");
58  
59    private static HRegionServer badRS;
60  
61    /**
62     * @throws java.lang.Exception
63     */
64    @BeforeClass
65    public static void setUpBeforeClass() throws Exception {
66      Configuration conf = TEST_UTIL.getConfiguration();
67      conf.setStrings(HConstants.REGION_SERVER_IMPL,
68          RegionServerWithFakeRpcServices.class.getName());
69      TEST_UTIL.startMiniCluster(1);
70      TEST_UTIL.getHBaseCluster().waitForActiveAndReadyMaster();
71      TEST_UTIL.waitUntilAllRegionsAssigned(TABLE_NAME.META_TABLE_NAME);
72      badRS = TEST_UTIL.getHBaseCluster().getRegionServer(0);
73      assertTrue(badRS.getRSRpcServices() instanceof FakeRSRpcServices);
74      HTableDescriptor table = new HTableDescriptor(TABLE_NAME);
75      HColumnDescriptor fam = new HColumnDescriptor(FAMILY);
76      fam.setMaxVersions(2);
77      table.addFamily(fam);
78      TEST_UTIL.createTable(table, null);
79    }
80  
81  
82    /**
83     * @throws java.lang.Exception
84     */
85    @AfterClass
86    public static void tearDownAfterClass() throws Exception {
87      TEST_UTIL.shutdownMiniCluster();
88    }
89  
90    @Test
91    public void testPreserveMetaCacheOnException() throws Exception {
92      ((FakeRSRpcServices)badRS.getRSRpcServices()).setExceptionInjector(
93          new RoundRobinExceptionInjector());
94      Configuration conf = new Configuration(TEST_UTIL.getConfiguration());
95      conf.set("hbase.client.retries.number", "1");
96      ConnectionManager.HConnectionImplementation conn =
97          (ConnectionManager.HConnectionImplementation) ConnectionFactory.createConnection(conf);
98      try {
99        Table table = conn.getTable(TABLE_NAME);
100       byte[] row = Bytes.toBytes("row1");
101 
102       Put put = new Put(row);
103       put.addColumn(FAMILY, QUALIFIER, Bytes.toBytes(10));
104       Get get = new Get(row);
105       Append append = new Append(row);
106       append.add(FAMILY, QUALIFIER, Bytes.toBytes(11));
107       Increment increment = new Increment(row);
108       increment.addColumn(FAMILY, QUALIFIER, 10);
109       Delete delete = new Delete(row);
110       delete.addColumn(FAMILY, QUALIFIER);
111       RowMutations mutations = new RowMutations(row);
112       mutations.add(put);
113       mutations.add(delete);
114 
115       Exception exp;
116       boolean success;
117       for (int i = 0; i < 50; i++) {
118         exp = null;
119         success = false;
120         try {
121           table.put(put);
122           // If at least one operation succeeded, we should have cached the region location.
123           success = true;
124           table.get(get);
125           table.append(append);
126           table.increment(increment);
127           table.delete(delete);
128           table.mutateRow(mutations);
129         } catch (IOException ex) {
130           // Only keep track of the last exception that updated the meta cache
131           if (ClientExceptionsUtil.isMetaClearingException(ex) || success) {
132             exp = ex;
133           }
134         }
135         // Do not test if we did not touch the meta cache in this iteration.
136         if (exp != null && ClientExceptionsUtil.isMetaClearingException(exp)) {
137           assertNull(conn.getCachedLocation(TABLE_NAME, row));
138         } else if (success) {
139           assertNotNull(conn.getCachedLocation(TABLE_NAME, row));
140         }
141       }
142     } finally {
143       conn.close();
144     }
145   }
146 
147   @Test
148   public void testCacheClearingOnCallQueueTooBig() throws Exception {
149     ((FakeRSRpcServices)badRS.getRSRpcServices()).setExceptionInjector(
150         new CallQueueTooBigExceptionInjector());
151     Configuration conf = new Configuration(TEST_UTIL.getConfiguration());
152     conf.set("hbase.client.retries.number", "2");
153     conf.set(MetricsConnection.CLIENT_SIDE_METRICS_ENABLED_KEY, "true");
154     ConnectionManager.HConnectionImplementation conn =
155         (ConnectionManager.HConnectionImplementation) ConnectionFactory.createConnection(conf);
156     try {
157       Table table = conn.getTable(TABLE_NAME);
158       byte[] row = Bytes.toBytes("row1");
159 
160       Put put = new Put(row);
161       put.addColumn(FAMILY, QUALIFIER, Bytes.toBytes(10));
162       table.put(put);
163 
164       // obtain the client metrics
165       MetricsConnection metrics = conn.getConnectionMetrics();
166       long preGetRegionClears = metrics.metaCacheNumClearRegion.count();
167       long preGetServerClears = metrics.metaCacheNumClearServer.count();
168 
169       // attempt a get on the test table
170       Get get = new Get(row);
171       try {
172         table.get(get);
173         fail("Expected CallQueueTooBigException");
174       } catch (RetriesExhaustedException ree) {
175         // expected
176       }
177 
178       // verify that no cache clearing took place
179       long postGetRegionClears = metrics.metaCacheNumClearRegion.count();
180       long postGetServerClears = metrics.metaCacheNumClearServer.count();
181       assertEquals(preGetRegionClears, postGetRegionClears);
182       assertEquals(preGetServerClears, postGetServerClears);
183     } finally {
184       conn.close();
185     }
186   }
187 
188   public static List<Throwable> metaCachePreservingExceptions() {
189     return new ArrayList<Throwable>() {{
190       add(new RegionOpeningException(" "));
191       add(new RegionTooBusyException());
192       add(new ThrottlingException(" "));
193       add(new MultiActionResultTooLarge(" "));
194       add(new RetryImmediatelyException(" "));
195       add(new CallQueueTooBigException());
196     }};
197   }
198 
199   public static class RegionServerWithFakeRpcServices extends HRegionServer {
200     private FakeRSRpcServices rsRpcServices;
201 
202     public RegionServerWithFakeRpcServices(Configuration conf, CoordinatedStateManager cp)
203       throws IOException, InterruptedException {
204       super(conf, cp);
205     }
206 
207     @Override
208     protected RSRpcServices createRpcServices() throws IOException {
209       this.rsRpcServices = new FakeRSRpcServices(this);
210       return rsRpcServices;
211     }
212 
213     public void setExceptionInjector(ExceptionInjector injector) {
214       rsRpcServices.setExceptionInjector(injector);
215     }
216   }
217 
218   public static class FakeRSRpcServices extends RSRpcServices {
219 
220     private ExceptionInjector exceptions;
221 
222     public FakeRSRpcServices(HRegionServer rs) throws IOException {
223       super(rs);
224       exceptions = new RoundRobinExceptionInjector();
225     }
226 
227     public void setExceptionInjector(ExceptionInjector injector) {
228       this.exceptions = injector;
229     }
230 
231     @Override
232     public GetResponse get(final RpcController controller,
233                            final ClientProtos.GetRequest request) throws ServiceException {
234       exceptions.throwOnGet(this, request);
235       return super.get(controller, request);
236     }
237 
238     @Override
239     public ClientProtos.MutateResponse mutate(final RpcController controller,
240         final ClientProtos.MutateRequest request) throws ServiceException {
241       exceptions.throwOnMutate(this, request);
242       return super.mutate(controller, request);
243     }
244 
245     @Override
246     public ClientProtos.ScanResponse scan(final RpcController controller,
247         final ClientProtos.ScanRequest request) throws ServiceException {
248       exceptions.throwOnScan(this, request);
249       return super.scan(controller, request);
250     }
251 
252     public Region getRegion(
253         final HBaseProtos.RegionSpecifier regionSpecifier) throws IOException {
254       return super.getRegion(regionSpecifier);
255     }
256   }
257 
258   public static abstract class ExceptionInjector {
259     protected boolean isTestTable(FakeRSRpcServices rpcServices,
260                                   HBaseProtos.RegionSpecifier regionSpec) throws ServiceException {
261       try {
262         return TABLE_NAME.equals(
263             rpcServices.getRegion(regionSpec).getTableDesc().getTableName());
264       } catch (IOException ioe) {
265         throw new ServiceException(ioe);
266       }
267     }
268 
269     public abstract void throwOnGet(FakeRSRpcServices rpcServices, ClientProtos.GetRequest request)
270         throws ServiceException;
271 
272     public abstract void throwOnMutate(FakeRSRpcServices rpcServices, ClientProtos.MutateRequest request)
273         throws ServiceException;
274 
275     public abstract void throwOnScan(FakeRSRpcServices rpcServices, ClientProtos.ScanRequest request)
276         throws ServiceException;
277   }
278 
279   /**
280    * Rotates through the possible cache clearing and non-cache clearing exceptions
281    * for requests.
282    */
283   public static class RoundRobinExceptionInjector extends ExceptionInjector {
284     private int numReqs = -1;
285     private int expCount = -1;
286     private List<Throwable> metaCachePreservingExceptions = metaCachePreservingExceptions();
287 
288     public void throwOnGet(FakeRSRpcServices rpcServices, ClientProtos.GetRequest request)
289         throws ServiceException {
290       throwSomeExceptions(rpcServices, request.getRegion());
291     }
292 
293     public void throwOnMutate(FakeRSRpcServices rpcServices, ClientProtos.MutateRequest request)
294         throws ServiceException {
295       throwSomeExceptions(rpcServices, request.getRegion());
296     }
297 
298     public void throwOnScan(FakeRSRpcServices rpcServices, ClientProtos.ScanRequest request)
299         throws ServiceException {
300       if (!request.hasScannerId()) {
301         // only handle initial scan requests
302         throwSomeExceptions(rpcServices, request.getRegion());
303       }
304     }
305 
306     /**
307      * Throw some exceptions. Mostly throw exceptions which do not clear meta cache.
308      * Periodically throw NotSevingRegionException which clears the meta cache.
309      * @throws ServiceException
310      */
311     private void throwSomeExceptions(FakeRSRpcServices rpcServices,
312                                      HBaseProtos.RegionSpecifier regionSpec)
313         throws ServiceException {
314       if (!isTestTable(rpcServices, regionSpec)) {
315         return;
316       }
317 
318       numReqs++;
319       // Succeed every 5 request, throw cache clearing exceptions twice every 5 requests and throw
320       // meta cache preserving exceptions otherwise.
321       if (numReqs % 5 ==0) {
322         return;
323       } else if (numReqs % 5 == 1 || numReqs % 5 == 2) {
324         throw new ServiceException(new NotServingRegionException());
325       }
326       // Round robin between different special exceptions.
327       // This is not ideal since exception types are not tied to the operation performed here,
328       // But, we don't really care here if we throw MultiActionTooLargeException while doing
329       // single Gets.
330       expCount++;
331       Throwable t = metaCachePreservingExceptions.get(
332           expCount % metaCachePreservingExceptions.size());
333       throw new ServiceException(t);
334     }
335   }
336 
337   /**
338    * Throws CallQueueTooBigException for all gets.
339    */
340   public static class CallQueueTooBigExceptionInjector extends ExceptionInjector {
341     @Override
342     public void throwOnGet(FakeRSRpcServices rpcServices, ClientProtos.GetRequest request)
343         throws ServiceException {
344       if (isTestTable(rpcServices, request.getRegion())) {
345         throw new ServiceException(new CallQueueTooBigException());
346       }
347     }
348 
349     @Override
350     public void throwOnMutate(FakeRSRpcServices rpcServices, ClientProtos.MutateRequest request)
351         throws ServiceException {
352     }
353 
354     @Override
355     public void throwOnScan(FakeRSRpcServices rpcServices, ClientProtos.ScanRequest request)
356         throws ServiceException {
357     }
358   }
359 }