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 java.io.IOException;
21  import java.util.concurrent.atomic.AtomicLong;
22  
23  import org.apache.hadoop.conf.Configuration;
24  import org.apache.hadoop.hbase.HConstants;
25  import org.apache.hadoop.hbase.RegionLocations;
26  import org.apache.hadoop.hbase.TableName;
27  import org.apache.hadoop.hbase.HRegionInfo;
28  import org.apache.hadoop.hbase.HRegionLocation;
29  import org.apache.hadoop.hbase.ServerName;
30  import org.apache.hadoop.hbase.ZooKeeperConnectionException;
31  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos;
32  import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
33  import org.apache.hadoop.hbase.util.Threads;
34  import org.apache.hadoop.hbase.client.ConnectionManager.HConnectionImplementation;
35  import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
36  import org.apache.hadoop.hbase.coprocessor.ObserverContext;
37  import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
38  import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
39  import org.mockito.Mockito;
40  import org.mockito.invocation.InvocationOnMock;
41  import org.mockito.stubbing.Answer;
42  
43  /**
44   * {@link ClusterConnection} testing utility.
45   */
46  public class HConnectionTestingUtility {
47    /*
48     * Not part of {@link HBaseTestingUtility} because this class is not
49     * in same package as {@link HConnection}.  Would have to reveal ugly
50     * {@link HConnectionManager} innards to HBaseTestingUtility to give it access.
51     */
52    /**
53     * Get a Mocked {@link HConnection} that goes with the passed <code>conf</code>
54     * configuration instance.  Minimally the mock will return
55     * <code>conf</conf> when {@link ClusterConnection#getConfiguration()} is invoked.
56     * Be sure to shutdown the connection when done by calling
57     * {@link HConnectionManager#deleteConnection(Configuration)} else it
58     * will stick around; this is probably not what you want.
59     * @param conf configuration
60     * @return HConnection object for <code>conf</code>
61     * @throws ZooKeeperConnectionException
62     */
63    public static ClusterConnection getMockedConnection(final Configuration conf)
64    throws ZooKeeperConnectionException {
65      HConnectionKey connectionKey = new HConnectionKey(conf);
66      synchronized (ConnectionManager.CONNECTION_INSTANCES) {
67        HConnectionImplementation connection =
68            ConnectionManager.CONNECTION_INSTANCES.get(connectionKey);
69        if (connection == null) {
70          connection = Mockito.mock(HConnectionImplementation.class);
71          Mockito.when(connection.getConfiguration()).thenReturn(conf);
72          Mockito.when(connection.getRpcControllerFactory()).thenReturn(
73          Mockito.mock(RpcControllerFactory.class));
74          // we need a real retrying caller
75          RpcRetryingCallerFactory callerFactory = new RpcRetryingCallerFactory(conf);
76          Mockito.when(connection.getRpcRetryingCallerFactory()).thenReturn(callerFactory);
77          ConnectionManager.CONNECTION_INSTANCES.put(connectionKey, connection);
78        }
79        return connection;
80      }
81    }
82  
83    /**
84     * @param connection
85     */
86    private static void mockRegionLocator(final HConnectionImplementation connection) {
87      try {
88        Mockito.when(connection.getRegionLocator(Mockito.any(TableName.class))).thenAnswer(
89            new Answer<RegionLocator>() {
90              @Override
91              public RegionLocator answer(InvocationOnMock invocation) throws Throwable {
92                TableName tableName = (TableName) invocation.getArguments()[0];
93                return new HRegionLocator(tableName, connection);
94              }
95            });
96      } catch (IOException e) {
97      }
98    }
99  
100   /**
101    * Calls {@link #getMockedConnection(Configuration)} and then mocks a few
102    * more of the popular {@link ClusterConnection} methods so they do 'normal'
103    * operation (see return doc below for list). Be sure to shutdown the
104    * connection when done by calling
105    * {@link HConnectionManager#deleteConnection(Configuration)} else it
106    * will stick around; this is probably not what you want.
107    *
108    * @param conf Configuration to use
109    * @param admin An AdminProtocol; can be null but is usually
110    * itself a mock.
111    * @param client A ClientProtocol; can be null but is usually
112    * itself a mock.
113    * @param sn ServerName to include in the region location returned by this
114    * <code>connection</code>
115    * @param hri HRegionInfo to include in the location returned when
116    * getRegionLocator is called on the mocked connection
117    * @return Mock up a connection that returns a {@link Configuration} when
118    * {@link ClusterConnection#getConfiguration()} is called, a 'location' when
119    * {@link ClusterConnection#getRegionLocation(org.apache.hadoop.hbase.TableName, byte[], boolean)}
120    * is called,
121    * and that returns the passed {@link AdminProtos.AdminService.BlockingInterface} instance when
122    * {@link ClusterConnection#getAdmin(ServerName)} is called, returns the passed
123    * {@link ClientProtos.ClientService.BlockingInterface} instance when
124    * {@link ClusterConnection#getClient(ServerName)} is called (Be sure to call
125    * {@link HConnectionManager#deleteConnection(Configuration)}
126    * when done with this mocked Connection.
127    * @throws IOException
128    */
129   public static ClusterConnection getMockedConnectionAndDecorate(final Configuration conf,
130       final AdminProtos.AdminService.BlockingInterface admin,
131       final ClientProtos.ClientService.BlockingInterface client,
132       final ServerName sn, final HRegionInfo hri)
133   throws IOException {
134     HConnectionImplementation c = Mockito.mock(HConnectionImplementation.class);
135     Mockito.when(c.getConfiguration()).thenReturn(conf);
136     ConnectionManager.CONNECTION_INSTANCES.put(new HConnectionKey(conf), c);
137     Mockito.doNothing().when(c).close();
138     // Make it so we return a particular location when asked.
139     final HRegionLocation loc = new HRegionLocation(hri, sn);
140     mockRegionLocator(c);
141     Mockito.when(c.getRegionLocation((TableName) Mockito.any(),
142         (byte[]) Mockito.any(), Mockito.anyBoolean())).
143       thenReturn(loc);
144     Mockito.when(c.locateRegion((TableName) Mockito.any(), (byte[]) Mockito.any())).
145       thenReturn(loc);
146     Mockito.when(c.locateRegion((TableName) Mockito.any(), (byte[]) Mockito.any(),
147         Mockito.anyBoolean(), Mockito.anyBoolean(),  Mockito.anyInt()))
148         .thenReturn(new RegionLocations(loc));
149     if (admin != null) {
150       // If a call to getAdmin, return this implementation.
151       Mockito.when(c.getAdmin(Mockito.any(ServerName.class))).
152         thenReturn(admin);
153     }
154     if (client != null) {
155       // If a call to getClient, return this client.
156       Mockito.when(c.getClient(Mockito.any(ServerName.class))).
157         thenReturn(client);
158     }
159     NonceGenerator ng = Mockito.mock(NonceGenerator.class);
160     Mockito.when(c.getNonceGenerator()).thenReturn(ng);
161     Mockito.when(c.getAsyncProcess()).thenReturn(
162       new AsyncProcess(c, conf, null, RpcRetryingCallerFactory.instantiate(conf), false,
163           RpcControllerFactory.instantiate(conf),
164           conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
165               HConstants.DEFAULT_HBASE_RPC_TIMEOUT)));
166     Mockito.doNothing().when(c).incCount();
167     Mockito.doNothing().when(c).decCount();
168     Mockito.when(c.getNewRpcRetryingCallerFactory(conf)).thenReturn(
169         RpcRetryingCallerFactory.instantiate(conf,
170             RetryingCallerInterceptorFactory.NO_OP_INTERCEPTOR, null));
171     Mockito.when(c.getRpcControllerFactory()).thenReturn(Mockito.mock(RpcControllerFactory.class));
172     HTableInterface t = Mockito.mock(HTableInterface.class);
173     Mockito.when(c.getTable((TableName)Mockito.any())).thenReturn(t);
174     ResultScanner rs = Mockito.mock(ResultScanner.class);
175     Mockito.when(t.getScanner((Scan)Mockito.any())).thenReturn(rs);
176     return c;
177   }
178 
179   /**
180    * Get a Mockito spied-upon {@link ClusterConnection} that goes with the passed
181    * <code>conf</code> configuration instance.
182    * Be sure to shutdown the connection when done by calling
183    * {@link HConnectionManager#deleteConnection(Configuration)} else it
184    * will stick around; this is probably not what you want.
185    * @param conf configuration
186    * @return HConnection object for <code>conf</code>
187    * @throws ZooKeeperConnectionException
188    * @see @link
189    * {http://mockito.googlecode.com/svn/branches/1.6/javadoc/org/mockito/Mockito.html#spy(T)}
190    */
191   public static ClusterConnection getSpiedConnection(final Configuration conf)
192   throws IOException {
193     HConnectionKey connectionKey = new HConnectionKey(conf);
194     synchronized (ConnectionManager.CONNECTION_INSTANCES) {
195       HConnectionImplementation connection =
196           ConnectionManager.CONNECTION_INSTANCES.get(connectionKey);
197       if (connection == null) {
198         connection = Mockito.spy(new HConnectionImplementation(conf, true));
199         ConnectionManager.CONNECTION_INSTANCES.put(connectionKey, connection);
200       }
201       return connection;
202     }
203   }
204 
205   public static ClusterConnection getSpiedClusterConnection(final Configuration conf)
206   throws IOException {
207     HConnectionKey connectionKey = new HConnectionKey(conf);
208     synchronized (ConnectionManager.CONNECTION_INSTANCES) {
209       HConnectionImplementation connection =
210           ConnectionManager.CONNECTION_INSTANCES.get(connectionKey);
211       if (connection == null) {
212         connection = Mockito.spy(new HConnectionImplementation(conf, true));
213         ConnectionManager.CONNECTION_INSTANCES.put(connectionKey, connection);
214       }
215       return connection;
216     }
217   }
218 
219   /**
220    * @return Count of extant connection instances
221    */
222   public static int getConnectionCount() {
223     synchronized (ConnectionManager.CONNECTION_INSTANCES) {
224       return ConnectionManager.CONNECTION_INSTANCES.size();
225     }
226   }
227 
228   /**
229    * This coproceesor sleep 2s at first increment/append rpc call.
230    */
231   public static class SleepAtFirstRpcCall extends BaseRegionObserver {
232     static final AtomicLong ct = new AtomicLong(0);
233     static final String SLEEP_TIME_CONF_KEY =
234         "hbase.coprocessor.SleepAtFirstRpcCall.sleepTime";
235     static final long DEFAULT_SLEEP_TIME = 2000;
236     static final AtomicLong sleepTime = new AtomicLong(DEFAULT_SLEEP_TIME);
237 
238     public SleepAtFirstRpcCall() {
239     }
240 
241     @Override
242     public void postOpen(ObserverContext<RegionCoprocessorEnvironment> e) {
243       RegionCoprocessorEnvironment env = e.getEnvironment();
244       Configuration conf = env.getConfiguration();
245       sleepTime.set(conf.getLong(SLEEP_TIME_CONF_KEY, DEFAULT_SLEEP_TIME));
246     }
247 
248     @Override
249     public Result postIncrement(final ObserverContext<RegionCoprocessorEnvironment> e,
250         final Increment increment, final Result result) throws IOException {
251       if (ct.incrementAndGet() == 1) {
252         Threads.sleep(sleepTime.get());
253       }
254       return result;
255     }
256 
257     @Override
258     public Result postAppend(final ObserverContext<RegionCoprocessorEnvironment> e,
259         final Append append, final Result result) throws IOException {
260       if (ct.incrementAndGet() == 1) {
261         Threads.sleep(sleepTime.get());
262       }
263       return result;
264     }
265   }
266 
267 }