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.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
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),
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
181
182
183
184
185
186
187
188
189
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
221
222 public static int getConnectionCount() {
223 synchronized (ConnectionManager.CONNECTION_INSTANCES) {
224 return ConnectionManager.CONNECTION_INSTANCES.size();
225 }
226 }
227
228
229
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 }