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 static org.apache.hadoop.hbase.HConstants.EMPTY_END_ROW;
21 import static org.apache.hadoop.hbase.HConstants.EMPTY_START_ROW;
22
23 import com.google.common.annotations.VisibleForTesting;
24
25 import java.io.IOException;
26 import java.util.Arrays;
27 import java.util.Comparator;
28 import java.util.List;
29 import java.util.concurrent.ExecutorService;
30 import java.util.concurrent.ThreadLocalRandom;
31
32 import org.apache.commons.logging.Log;
33 import org.apache.hadoop.conf.Configuration;
34 import org.apache.hadoop.hbase.Cell;
35 import org.apache.hadoop.hbase.CellComparator;
36 import org.apache.hadoop.hbase.CellUtil;
37 import org.apache.hadoop.hbase.HConstants;
38 import org.apache.hadoop.hbase.HRegionInfo;
39 import org.apache.hadoop.hbase.ServerName;
40 import org.apache.hadoop.hbase.TableName;
41 import org.apache.hadoop.hbase.classification.InterfaceAudience;
42 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService;
43 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService;
44 import org.apache.hadoop.hbase.security.User;
45 import org.apache.hadoop.hbase.security.UserProvider;
46 import org.apache.hadoop.hbase.util.Bytes;
47
48
49
50
51 @InterfaceAudience.Private
52 public class ConnectionUtils {
53
54
55
56
57
58
59
60
61 public static long getPauseTime(final long pause, final int tries) {
62 int ntries = tries;
63 if (ntries >= HConstants.RETRY_BACKOFF.length) {
64 ntries = HConstants.RETRY_BACKOFF.length - 1;
65 }
66 if (ntries < 0) {
67 ntries = 0;
68 }
69
70 long normalPause = pause * HConstants.RETRY_BACKOFF[ntries];
71
72 long jitter = (long) (normalPause * ThreadLocalRandom.current().nextFloat() * 0.01f);
73 return normalPause + jitter;
74 }
75
76
77
78
79
80
81
82 public static NonceGenerator injectNonceGeneratorForTesting(
83 ClusterConnection conn, NonceGenerator cnm) {
84 return ConnectionManager.injectNonceGeneratorForTesting(conn, cnm);
85 }
86
87
88
89
90
91
92
93
94 public static void setServerSideHConnectionRetriesConfig(
95 final Configuration c, final String sn, final Log log) {
96
97 int hcRetries = c.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
98 HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
99
100
101 int serversideMultiplier = c.getInt("hbase.client.serverside.retries.multiplier", 10);
102 int retries = hcRetries * serversideMultiplier;
103 c.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, retries);
104 log.info(sn + " server-side HConnection retries=" + retries);
105 }
106
107
108
109
110
111
112
113
114
115
116 @Deprecated
117 public static ClusterConnection createShortCircuitHConnection(final Connection conn,
118 final ServerName serverName, final AdminService.BlockingInterface admin,
119 final ClientService.BlockingInterface client) {
120 return new ConnectionAdapter(conn) {
121 @Override
122 public AdminService.BlockingInterface getAdmin(
123 ServerName sn, boolean getMaster) throws IOException {
124 return serverName.equals(sn) ? admin : super.getAdmin(sn, getMaster);
125 }
126
127 @Override
128 public ClientService.BlockingInterface getClient(
129 ServerName sn) throws IOException {
130 return serverName.equals(sn) ? client : super.getClient(sn);
131 }
132 };
133 }
134
135
136
137
138
139
140
141
142
143
144
145
146
147 public static ClusterConnection createShortCircuitConnection(final Configuration conf,
148 ExecutorService pool, User user, final ServerName serverName,
149 final AdminService.BlockingInterface admin, final ClientService.BlockingInterface client)
150 throws IOException {
151 if (user == null) {
152 user = UserProvider.instantiate(conf).getCurrent();
153 }
154 return new ConnectionManager.HConnectionImplementation(conf, false, pool, user) {
155 @Override
156 public AdminService.BlockingInterface getAdmin(ServerName sn, boolean getMaster)
157 throws IOException {
158 return serverName.equals(sn) ? admin : super.getAdmin(sn, getMaster);
159 }
160
161 @Override
162 public ClientService.BlockingInterface getClient(ServerName sn) throws IOException {
163 return serverName.equals(sn) ? client : super.getClient(sn);
164 }
165 };
166 }
167
168
169
170
171
172 @VisibleForTesting
173 public static void setupMasterlessConnection(Configuration conf) {
174 conf.set(HConnection.HBASE_CLIENT_CONNECTION_IMPL,
175 MasterlessConnection.class.getName());
176 }
177
178
179
180
181
182 static class MasterlessConnection extends ConnectionManager.HConnectionImplementation {
183 MasterlessConnection(Configuration conf, boolean managed,
184 ExecutorService pool, User user) throws IOException {
185 super(conf, managed, pool, user);
186 }
187
188 @Override
189 public boolean isTableDisabled(TableName tableName) throws IOException {
190
191 return false;
192 }
193 }
194
195
196
197 static final byte[] MAX_BYTE_ARRAY = Bytes.createMaxByteArray(9);
198
199
200
201
202 static byte[] createClosestRowAfter(byte[] row) {
203 return Arrays.copyOf(row, row.length + 1);
204 }
205
206
207
208
209 static byte[] createCloseRowBefore(byte[] row) {
210 if (row.length == 0) {
211 return MAX_BYTE_ARRAY;
212 }
213 if (row[row.length - 1] == 0) {
214 return Arrays.copyOf(row, row.length - 1);
215 } else {
216 byte[] nextRow = new byte[row.length + MAX_BYTE_ARRAY.length];
217 System.arraycopy(row, 0, nextRow, 0, row.length - 1);
218 nextRow[row.length - 1] = (byte) ((row[row.length - 1] & 0xFF) - 1);
219 System.arraycopy(MAX_BYTE_ARRAY, 0, nextRow, row.length, MAX_BYTE_ARRAY.length);
220 return nextRow;
221 }
222 }
223
224 static boolean isEmptyStartRow(byte[] row) {
225 return Bytes.equals(row, EMPTY_START_ROW);
226 }
227
228 static boolean isEmptyStopRow(byte[] row) {
229 return Bytes.equals(row, EMPTY_END_ROW);
230 }
231
232 private static final Comparator<Cell> COMPARE_WITHOUT_ROW = new Comparator<Cell>() {
233
234 @Override
235 public int compare(Cell o1, Cell o2) {
236 return CellComparator.compareWithoutRow(o1, o2);
237 }
238 };
239
240 static Result filterCells(Result result, Cell keepCellsAfter) {
241 if (keepCellsAfter == null) {
242
243 return result;
244 }
245
246 if (!CellUtil.matchingRow(keepCellsAfter, result.getRow(), 0, result.getRow().length)) {
247 return result;
248 }
249 Cell[] rawCells = result.rawCells();
250 int index = Arrays.binarySearch(rawCells, keepCellsAfter, COMPARE_WITHOUT_ROW);
251 if (index < 0) {
252 index = -index - 1;
253 } else {
254 index++;
255 }
256 if (index == 0) {
257 return result;
258 }
259 if (index == rawCells.length) {
260 return null;
261 }
262 return Result.create(Arrays.copyOfRange(rawCells, index, rawCells.length), null,
263 result.isStale(), result.mayHaveMoreCellsInRow());
264 }
265
266 static boolean noMoreResultsForScan(Scan scan, HRegionInfo info) {
267 if (isEmptyStopRow(info.getEndKey())) {
268 return true;
269 }
270 if (isEmptyStopRow(scan.getStopRow())) {
271 return false;
272 }
273 int c = Bytes.compareTo(info.getEndKey(), scan.getStopRow());
274
275
276
277 return c > 0 || (c == 0 && !scan.includeStopRow());
278 }
279
280 static boolean noMoreResultsForReverseScan(Scan scan, HRegionInfo info) {
281 if (isEmptyStartRow(info.getStartKey())) {
282 return true;
283 }
284 if (isEmptyStopRow(scan.getStopRow())) {
285 return false;
286 }
287
288
289 return Bytes.compareTo(info.getStartKey(), scan.getStopRow()) <= 0;
290 }
291
292 public static ScanResultCache createScanResultCache(Scan scan, List<Result> cache) {
293 if (scan.getAllowPartialResults()) {
294 return new AllowPartialScanResultCache(cache);
295 } else if (scan.getBatch() > 0) {
296 return new BatchScanResultCache(cache, scan.getBatch());
297 } else {
298 return new CompleteScanResultCache(cache);
299 }
300 }
301 }