1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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
45
46 public class HConnectionTestingUtility {
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
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
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
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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
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
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
151 Mockito.when(c.getAdmin(Mockito.any(ServerName.class))).
152 thenReturn(admin);
153 }
154 if (client != null) {
155
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
180
181
182
183
184
185
186
187
188
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
220
221 public static int getConnectionCount() {
222 synchronized (ConnectionManager.CONNECTION_INSTANCES) {
223 return ConnectionManager.CONNECTION_INSTANCES.size();
224 }
225 }
226
227
228
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 }