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.RegionLocations;
25  import org.apache.hadoop.hbase.TableName;
26  import org.apache.hadoop.hbase.HConstants;
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), conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
164           HConstants.DEFAULT_HBASE_RPC_TIMEOUT)));
165     Mockito.doNothing().when(c).incCount();
166     Mockito.doNothing().when(c).decCount();
167     Mockito.when(c.getNewRpcRetryingCallerFactory(conf)).thenReturn(
168         RpcRetryingCallerFactory.instantiate(conf,
169             RetryingCallerInterceptorFactory.NO_OP_INTERCEPTOR, null));
170     Mockito.when(c.getRpcControllerFactory()).thenReturn(Mockito.mock(RpcControllerFactory.class));
171     HTableInterface t = Mockito.mock(HTableInterface.class);
172     Mockito.when(c.getTable((TableName)Mockito.any())).thenReturn(t);
173     ResultScanner rs = Mockito.mock(ResultScanner.class);
174     Mockito.when(t.getScanner((Scan)Mockito.any())).thenReturn(rs);
175     return c;
176   }
177 
178   /**
179    * Get a Mockito spied-upon {@link ClusterConnection} that goes with the passed
180    * <code>conf</code> configuration instance.
181    * Be sure to shutdown the connection when done by calling
182    * {@link HConnectionManager#deleteConnection(Configuration)} else it
183    * will stick around; this is probably not what you want.
184    * @param conf configuration
185    * @return HConnection object for <code>conf</code>
186    * @throws ZooKeeperConnectionException
187    * @see @link
188    * {http://mockito.googlecode.com/svn/branches/1.6/javadoc/org/mockito/Mockito.html#spy(T)}
189    */
190   public static ClusterConnection getSpiedConnection(final Configuration conf)
191   throws IOException {
192     HConnectionKey connectionKey = new HConnectionKey(conf);
193     synchronized (ConnectionManager.CONNECTION_INSTANCES) {
194       HConnectionImplementation connection =
195           ConnectionManager.CONNECTION_INSTANCES.get(connectionKey);
196       if (connection == null) {
197         connection = Mockito.spy(new HConnectionImplementation(conf, true));
198         ConnectionManager.CONNECTION_INSTANCES.put(connectionKey, connection);
199       }
200       return connection;
201     }
202   }
203 
204   public static ClusterConnection getSpiedClusterConnection(final Configuration conf)
205   throws IOException {
206     HConnectionKey connectionKey = new HConnectionKey(conf);
207     synchronized (ConnectionManager.CONNECTION_INSTANCES) {
208       HConnectionImplementation connection =
209           ConnectionManager.CONNECTION_INSTANCES.get(connectionKey);
210       if (connection == null) {
211         connection = Mockito.spy(new HConnectionImplementation(conf, true));
212         ConnectionManager.CONNECTION_INSTANCES.put(connectionKey, connection);
213       }
214       return connection;
215     }
216   }
217 
218   /**
219    * @return Count of extant connection instances
220    */
221   public static int getConnectionCount() {
222     synchronized (ConnectionManager.CONNECTION_INSTANCES) {
223       return ConnectionManager.CONNECTION_INSTANCES.size();
224     }
225   }
226 
227   /**
228    * This coproceesor sleep 2s at first increment/append rpc call.
229    */
230   public static class SleepAtFirstRpcCall extends BaseRegionObserver {
231     static final AtomicLong ct = new AtomicLong(0);
232     static final String SLEEP_TIME_CONF_KEY =
233         "hbase.coprocessor.SleepAtFirstRpcCall.sleepTime";
234     static final long DEFAULT_SLEEP_TIME = 2000;
235     static final AtomicLong sleepTime = new AtomicLong(DEFAULT_SLEEP_TIME);
236 
237     public SleepAtFirstRpcCall() {
238     }
239 
240     @Override
241     public void postOpen(ObserverContext<RegionCoprocessorEnvironment> e) {
242       RegionCoprocessorEnvironment env = e.getEnvironment();
243       Configuration conf = env.getConfiguration();
244       sleepTime.set(conf.getLong(SLEEP_TIME_CONF_KEY, DEFAULT_SLEEP_TIME));
245     }
246 
247     @Override
248     public Result postIncrement(final ObserverContext<RegionCoprocessorEnvironment> e,
249         final Increment increment, final Result result) throws IOException {
250       if (ct.incrementAndGet() == 1) {
251         Threads.sleep(sleepTime.get());
252       }
253       return result;
254     }
255 
256     @Override
257     public Result postAppend(final ObserverContext<RegionCoprocessorEnvironment> e,
258         final Append append, final Result result) throws IOException {
259       if (ct.incrementAndGet() == 1) {
260         Threads.sleep(sleepTime.get());
261       }
262       return result;
263     }
264   }
265 
266 }