001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.client;
019
020import static java.util.stream.Collectors.toList;
021import static org.apache.hadoop.hbase.HConstants.EMPTY_END_ROW;
022import static org.apache.hadoop.hbase.HConstants.EMPTY_START_ROW;
023import static org.apache.hadoop.hbase.util.FutureUtils.addListener;
024
025import java.io.IOException;
026import java.lang.reflect.UndeclaredThrowableException;
027import java.net.InetAddress;
028import java.net.UnknownHostException;
029import java.util.Arrays;
030import java.util.List;
031import java.util.Optional;
032import java.util.concurrent.CompletableFuture;
033import java.util.concurrent.ExecutorService;
034import java.util.concurrent.ThreadLocalRandom;
035import java.util.concurrent.TimeUnit;
036import java.util.concurrent.atomic.AtomicReference;
037import java.util.function.Function;
038import java.util.function.Predicate;
039import java.util.function.Supplier;
040import org.apache.hadoop.conf.Configuration;
041import org.apache.hadoop.hbase.Cell;
042import org.apache.hadoop.hbase.CellComparator;
043import org.apache.hadoop.hbase.HConstants;
044import org.apache.hadoop.hbase.PrivateCellUtil;
045import org.apache.hadoop.hbase.RegionLocations;
046import org.apache.hadoop.hbase.ServerName;
047import org.apache.hadoop.hbase.TableName;
048import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
049import org.apache.hadoop.hbase.ipc.HBaseRpcController;
050import org.apache.hadoop.hbase.security.User;
051import org.apache.hadoop.hbase.security.UserProvider;
052import org.apache.hadoop.hbase.util.Bytes;
053import org.apache.hadoop.hbase.util.ReflectionUtils;
054import org.apache.hadoop.ipc.RemoteException;
055import org.apache.hadoop.net.DNS;
056import org.apache.yetus.audience.InterfaceAudience;
057import org.slf4j.Logger;
058import org.slf4j.LoggerFactory;
059
060import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
061import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
062import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException;
063import org.apache.hbase.thirdparty.io.netty.util.Timer;
064
065import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
066import org.apache.hadoop.hbase.shaded.protobuf.ResponseConverter;
067import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService;
068import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
069import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ClientService;
070import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ScanResponse;
071import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.MasterService;
072
073/**
074 * Utility used by client connections.
075 */
076@InterfaceAudience.Private
077public final class ConnectionUtils {
078
079  private static final Logger LOG = LoggerFactory.getLogger(ConnectionUtils.class);
080
081  private ConnectionUtils() {
082  }
083
084  /**
085   * Calculate pause time. Built on {@link HConstants#RETRY_BACKOFF}.
086   * @param pause time to pause
087   * @param tries amount of tries
088   * @return How long to wait after <code>tries</code> retries
089   */
090  public static long getPauseTime(final long pause, final int tries) {
091    int ntries = tries;
092    if (ntries >= HConstants.RETRY_BACKOFF.length) {
093      ntries = HConstants.RETRY_BACKOFF.length - 1;
094    }
095    if (ntries < 0) {
096      ntries = 0;
097    }
098
099    long normalPause = pause * HConstants.RETRY_BACKOFF[ntries];
100    // 1% possible jitter
101    long jitter = (long) (normalPause * ThreadLocalRandom.current().nextFloat() * 0.01f);
102    return normalPause + jitter;
103  }
104
105  /**
106   * @param conn The connection for which to replace the generator.
107   * @param cnm Replaces the nonce generator used, for testing.
108   * @return old nonce generator.
109   */
110  public static NonceGenerator injectNonceGeneratorForTesting(ClusterConnection conn,
111      NonceGenerator cnm) {
112    return ConnectionImplementation.injectNonceGeneratorForTesting(conn, cnm);
113  }
114
115  /**
116   * Changes the configuration to set the number of retries needed when using Connection internally,
117   * e.g. for updating catalog tables, etc. Call this method before we create any Connections.
118   * @param c The Configuration instance to set the retries into.
119   * @param log Used to log what we set in here.
120   */
121  public static void setServerSideHConnectionRetriesConfig(final Configuration c, final String sn,
122      final Logger log) {
123    // TODO: Fix this. Not all connections from server side should have 10 times the retries.
124    int hcRetries = c.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
125      HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
126    // Go big. Multiply by 10. If we can't get to meta after this many retries
127    // then something seriously wrong.
128    int serversideMultiplier = c.getInt(HConstants.HBASE_CLIENT_SERVERSIDE_RETRIES_MULTIPLIER,
129      HConstants.DEFAULT_HBASE_CLIENT_SERVERSIDE_RETRIES_MULTIPLIER);
130    int retries = hcRetries * serversideMultiplier;
131    c.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, retries);
132    log.info(sn + " server-side Connection retries=" + retries);
133  }
134
135  /**
136   * A ClusterConnection that will short-circuit RPC making direct invocations against the localhost
137   * if the invocation target is 'this' server; save on network and protobuf invocations.
138   */
139  // TODO This has to still do PB marshalling/unmarshalling stuff. Check how/whether we can avoid.
140  @VisibleForTesting // Class is visible so can assert we are short-circuiting when expected.
141  public static class ShortCircuitingClusterConnection extends ConnectionImplementation {
142    private final ServerName serverName;
143    private final AdminService.BlockingInterface localHostAdmin;
144    private final ClientService.BlockingInterface localHostClient;
145
146    private ShortCircuitingClusterConnection(Configuration conf, ExecutorService pool, User user,
147        ServerName serverName, AdminService.BlockingInterface admin,
148        ClientService.BlockingInterface client) throws IOException {
149      super(conf, pool, user);
150      this.serverName = serverName;
151      this.localHostAdmin = admin;
152      this.localHostClient = client;
153    }
154
155    @Override
156    public AdminService.BlockingInterface getAdmin(ServerName sn) throws IOException {
157      return serverName.equals(sn) ? this.localHostAdmin : super.getAdmin(sn);
158    }
159
160    @Override
161    public ClientService.BlockingInterface getClient(ServerName sn) throws IOException {
162      return serverName.equals(sn) ? this.localHostClient : super.getClient(sn);
163    }
164
165    @Override
166    public MasterKeepAliveConnection getMaster() throws IOException {
167      if (this.localHostClient instanceof MasterService.BlockingInterface) {
168        return new ShortCircuitMasterConnection(
169          (MasterService.BlockingInterface) this.localHostClient);
170      }
171      return super.getMaster();
172    }
173  }
174
175  /**
176   * Creates a short-circuit connection that can bypass the RPC layer (serialization,
177   * deserialization, networking, etc..) when talking to a local server.
178   * @param conf the current configuration
179   * @param pool the thread pool to use for batch operations
180   * @param user the user the connection is for
181   * @param serverName the local server name
182   * @param admin the admin interface of the local server
183   * @param client the client interface of the local server
184   * @return an short-circuit connection.
185   * @throws IOException if IO failure occurred
186   */
187  public static ClusterConnection createShortCircuitConnection(final Configuration conf,
188      ExecutorService pool, User user, final ServerName serverName,
189      final AdminService.BlockingInterface admin, final ClientService.BlockingInterface client)
190      throws IOException {
191    if (user == null) {
192      user = UserProvider.instantiate(conf).getCurrent();
193    }
194    return new ShortCircuitingClusterConnection(conf, pool, user, serverName, admin, client);
195  }
196
197  /**
198   * Setup the connection class, so that it will not depend on master being online. Used for testing
199   * @param conf configuration to set
200   */
201  @VisibleForTesting
202  public static void setupMasterlessConnection(Configuration conf) {
203    conf.set(ClusterConnection.HBASE_CLIENT_CONNECTION_IMPL, MasterlessConnection.class.getName());
204  }
205
206  /**
207   * Some tests shut down the master. But table availability is a master RPC which is performed on
208   * region re-lookups.
209   */
210  static class MasterlessConnection extends ConnectionImplementation {
211    MasterlessConnection(Configuration conf, ExecutorService pool, User user) throws IOException {
212      super(conf, pool, user);
213    }
214
215    @Override
216    public boolean isTableDisabled(TableName tableName) throws IOException {
217      // treat all tables as enabled
218      return false;
219    }
220  }
221
222  /**
223   * Return retires + 1. The returned value will be in range [1, Integer.MAX_VALUE].
224   */
225  static int retries2Attempts(int retries) {
226    return Math.max(1, retries == Integer.MAX_VALUE ? Integer.MAX_VALUE : retries + 1);
227  }
228
229  /**
230   * Get a unique key for the rpc stub to the given server.
231   */
232  static String getStubKey(String serviceName, ServerName serverName, boolean hostnameCanChange) {
233    // Sometimes, servers go down and they come back up with the same hostname but a different
234    // IP address. Force a resolution of the rsHostname by trying to instantiate an
235    // InetSocketAddress, and this way we will rightfully get a new stubKey.
236    // Also, include the hostname in the key so as to take care of those cases where the
237    // DNS name is different but IP address remains the same.
238    String hostname = serverName.getHostname();
239    int port = serverName.getPort();
240    if (hostnameCanChange) {
241      try {
242        InetAddress ip = InetAddress.getByName(hostname);
243        return serviceName + "@" + hostname + "-" + ip.getHostAddress() + ":" + port;
244      } catch (UnknownHostException e) {
245        LOG.warn("Can not resolve " + hostname + ", please check your network", e);
246      }
247    }
248    return serviceName + "@" + hostname + ":" + port;
249  }
250
251  static void checkHasFamilies(Mutation mutation) {
252    Preconditions.checkArgument(mutation.numFamilies() > 0,
253      "Invalid arguments to %s, zero columns specified", mutation.toString());
254  }
255
256  /** Dummy nonce generator for disabled nonces. */
257  static final NonceGenerator NO_NONCE_GENERATOR = new NonceGenerator() {
258
259    @Override
260    public long newNonce() {
261      return HConstants.NO_NONCE;
262    }
263
264    @Override
265    public long getNonceGroup() {
266      return HConstants.NO_NONCE;
267    }
268  };
269
270  // A byte array in which all elements are the max byte, and it is used to
271  // construct closest front row
272  static final byte[] MAX_BYTE_ARRAY = Bytes.createMaxByteArray(9);
273
274  /**
275   * Create the closest row after the specified row
276   */
277  static byte[] createClosestRowAfter(byte[] row) {
278    return Arrays.copyOf(row, row.length + 1);
279  }
280
281  /**
282   * Create a row before the specified row and very close to the specified row.
283   */
284  static byte[] createCloseRowBefore(byte[] row) {
285    if (row.length == 0) {
286      return MAX_BYTE_ARRAY;
287    }
288    if (row[row.length - 1] == 0) {
289      return Arrays.copyOf(row, row.length - 1);
290    } else {
291      byte[] nextRow = new byte[row.length + MAX_BYTE_ARRAY.length];
292      System.arraycopy(row, 0, nextRow, 0, row.length - 1);
293      nextRow[row.length - 1] = (byte) ((row[row.length - 1] & 0xFF) - 1);
294      System.arraycopy(MAX_BYTE_ARRAY, 0, nextRow, row.length, MAX_BYTE_ARRAY.length);
295      return nextRow;
296    }
297  }
298
299  static boolean isEmptyStartRow(byte[] row) {
300    return Bytes.equals(row, EMPTY_START_ROW);
301  }
302
303  static boolean isEmptyStopRow(byte[] row) {
304    return Bytes.equals(row, EMPTY_END_ROW);
305  }
306
307  static void resetController(HBaseRpcController controller, long timeoutNs, int priority) {
308    controller.reset();
309    if (timeoutNs >= 0) {
310      controller.setCallTimeout(
311        (int) Math.min(Integer.MAX_VALUE, TimeUnit.NANOSECONDS.toMillis(timeoutNs)));
312    }
313    controller.setPriority(priority);
314  }
315
316  static Throwable translateException(Throwable t) {
317    if (t instanceof UndeclaredThrowableException && t.getCause() != null) {
318      t = t.getCause();
319    }
320    if (t instanceof RemoteException) {
321      t = ((RemoteException) t).unwrapRemoteException();
322    }
323    if (t instanceof ServiceException && t.getCause() != null) {
324      t = translateException(t.getCause());
325    }
326    return t;
327  }
328
329  static long calcEstimatedSize(Result rs) {
330    long estimatedHeapSizeOfResult = 0;
331    // We don't make Iterator here
332    for (Cell cell : rs.rawCells()) {
333      estimatedHeapSizeOfResult += cell.heapSize();
334    }
335    return estimatedHeapSizeOfResult;
336  }
337
338  static Result filterCells(Result result, Cell keepCellsAfter) {
339    if (keepCellsAfter == null) {
340      // do not need to filter
341      return result;
342    }
343    // not the same row
344    if (!PrivateCellUtil.matchingRows(keepCellsAfter, result.getRow(), 0, result.getRow().length)) {
345      return result;
346    }
347    Cell[] rawCells = result.rawCells();
348    int index = Arrays.binarySearch(rawCells, keepCellsAfter,
349      CellComparator.getInstance()::compareWithoutRow);
350    if (index < 0) {
351      index = -index - 1;
352    } else {
353      index++;
354    }
355    if (index == 0) {
356      return result;
357    }
358    if (index == rawCells.length) {
359      return null;
360    }
361    return Result.create(Arrays.copyOfRange(rawCells, index, rawCells.length), null,
362      result.isStale(), result.mayHaveMoreCellsInRow());
363  }
364
365  // Add a delta to avoid timeout immediately after a retry sleeping.
366  static final long SLEEP_DELTA_NS = TimeUnit.MILLISECONDS.toNanos(1);
367
368  static Get toCheckExistenceOnly(Get get) {
369    if (get.isCheckExistenceOnly()) {
370      return get;
371    }
372    return ReflectionUtils.newInstance(get.getClass(), get).setCheckExistenceOnly(true);
373  }
374
375  static List<Get> toCheckExistenceOnly(List<Get> gets) {
376    return gets.stream().map(ConnectionUtils::toCheckExistenceOnly).collect(toList());
377  }
378
379  static RegionLocateType getLocateType(Scan scan) {
380    if (scan.isReversed()) {
381      if (isEmptyStartRow(scan.getStartRow())) {
382        return RegionLocateType.BEFORE;
383      } else {
384        return scan.includeStartRow() ? RegionLocateType.CURRENT : RegionLocateType.BEFORE;
385      }
386    } else {
387      return scan.includeStartRow() ? RegionLocateType.CURRENT : RegionLocateType.AFTER;
388    }
389  }
390
391  static boolean noMoreResultsForScan(Scan scan, RegionInfo info) {
392    if (isEmptyStopRow(info.getEndKey())) {
393      return true;
394    }
395    if (isEmptyStopRow(scan.getStopRow())) {
396      return false;
397    }
398    int c = Bytes.compareTo(info.getEndKey(), scan.getStopRow());
399    // 1. if our stop row is less than the endKey of the region
400    // 2. if our stop row is equal to the endKey of the region and we do not include the stop row
401    // for scan.
402    return c > 0 || (c == 0 && !scan.includeStopRow());
403  }
404
405  static boolean noMoreResultsForReverseScan(Scan scan, RegionInfo info) {
406    if (isEmptyStartRow(info.getStartKey())) {
407      return true;
408    }
409    if (isEmptyStopRow(scan.getStopRow())) {
410      return false;
411    }
412    // no need to test the inclusive of the stop row as the start key of a region is included in
413    // the region.
414    return Bytes.compareTo(info.getStartKey(), scan.getStopRow()) <= 0;
415  }
416
417  static <T> CompletableFuture<List<T>> allOf(List<CompletableFuture<T>> futures) {
418    return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
419      .thenApply(v -> futures.stream().map(f -> f.getNow(null)).collect(toList()));
420  }
421
422  public static ScanResultCache createScanResultCache(Scan scan) {
423    if (scan.getAllowPartialResults()) {
424      return new AllowPartialScanResultCache();
425    } else if (scan.getBatch() > 0) {
426      return new BatchScanResultCache(scan.getBatch());
427    } else {
428      return new CompleteScanResultCache();
429    }
430  }
431
432  private static final String MY_ADDRESS = getMyAddress();
433
434  private static String getMyAddress() {
435    try {
436      return DNS.getDefaultHost("default", "default");
437    } catch (UnknownHostException uhe) {
438      LOG.error("cannot determine my address", uhe);
439      return null;
440    }
441  }
442
443  static boolean isRemote(String host) {
444    return !host.equalsIgnoreCase(MY_ADDRESS);
445  }
446
447  static void incRPCCallsMetrics(ScanMetrics scanMetrics, boolean isRegionServerRemote) {
448    if (scanMetrics == null) {
449      return;
450    }
451    scanMetrics.countOfRPCcalls.incrementAndGet();
452    if (isRegionServerRemote) {
453      scanMetrics.countOfRemoteRPCcalls.incrementAndGet();
454    }
455  }
456
457  static void incRPCRetriesMetrics(ScanMetrics scanMetrics, boolean isRegionServerRemote) {
458    if (scanMetrics == null) {
459      return;
460    }
461    scanMetrics.countOfRPCRetries.incrementAndGet();
462    if (isRegionServerRemote) {
463      scanMetrics.countOfRemoteRPCRetries.incrementAndGet();
464    }
465  }
466
467  static void updateResultsMetrics(ScanMetrics scanMetrics, Result[] rrs,
468      boolean isRegionServerRemote) {
469    if (scanMetrics == null || rrs == null || rrs.length == 0) {
470      return;
471    }
472    long resultSize = 0;
473    for (Result rr : rrs) {
474      for (Cell cell : rr.rawCells()) {
475        resultSize += PrivateCellUtil.estimatedSerializedSizeOf(cell);
476      }
477    }
478    scanMetrics.countOfBytesInResults.addAndGet(resultSize);
479    if (isRegionServerRemote) {
480      scanMetrics.countOfBytesInRemoteResults.addAndGet(resultSize);
481    }
482  }
483
484  /**
485   * Use the scan metrics returned by the server to add to the identically named counters in the
486   * client side metrics. If a counter does not exist with the same name as the server side metric,
487   * the attempt to increase the counter will fail.
488   */
489  static void updateServerSideMetrics(ScanMetrics scanMetrics, ScanResponse response) {
490    if (scanMetrics == null || response == null || !response.hasScanMetrics()) {
491      return;
492    }
493    ResponseConverter.getScanMetrics(response).forEach(scanMetrics::addToCounter);
494  }
495
496  static void incRegionCountMetrics(ScanMetrics scanMetrics) {
497    if (scanMetrics == null) {
498      return;
499    }
500    scanMetrics.countOfRegions.incrementAndGet();
501  }
502
503  /**
504   * Connect the two futures, if the src future is done, then mark the dst future as done. And if
505   * the dst future is done, then cancel the src future. This is used for timeline consistent read.
506   * <p/>
507   * Pass empty metrics if you want to link the primary future and the dst future so we will not
508   * increase the hedge read related metrics.
509   */
510  private static <T> void connect(CompletableFuture<T> srcFuture, CompletableFuture<T> dstFuture,
511      Optional<MetricsConnection> metrics) {
512    addListener(srcFuture, (r, e) -> {
513      if (e != null) {
514        dstFuture.completeExceptionally(e);
515      } else {
516        if (dstFuture.complete(r)) {
517          metrics.ifPresent(MetricsConnection::incrHedgedReadWin);
518        }
519      }
520    });
521    // The cancellation may be a dummy one as the dstFuture may be completed by this srcFuture.
522    // Notice that this is a bit tricky, as the execution chain maybe 'complete src -> complete dst
523    // -> cancel src', for now it seems to be fine, as the will use CAS to set the result first in
524    // CompletableFuture. If later this causes problems, we could use whenCompleteAsync to break the
525    // tie.
526    addListener(dstFuture, (r, e) -> srcFuture.cancel(false));
527  }
528
529  private static <T> void sendRequestsToSecondaryReplicas(
530      Function<Integer, CompletableFuture<T>> requestReplica, RegionLocations locs,
531      CompletableFuture<T> future, Optional<MetricsConnection> metrics) {
532    if (future.isDone()) {
533      // do not send requests to secondary replicas if the future is done, i.e, the primary request
534      // has already been finished.
535      return;
536    }
537    for (int replicaId = 1, n = locs.size(); replicaId < n; replicaId++) {
538      CompletableFuture<T> secondaryFuture = requestReplica.apply(replicaId);
539      metrics.ifPresent(MetricsConnection::incrHedgedReadOps);
540      connect(secondaryFuture, future, metrics);
541    }
542  }
543
544  static <T> CompletableFuture<T> timelineConsistentRead(AsyncRegionLocator locator,
545      TableName tableName, Query query, byte[] row, RegionLocateType locateType,
546      Function<Integer, CompletableFuture<T>> requestReplica, long rpcTimeoutNs,
547      long primaryCallTimeoutNs, Timer retryTimer, Optional<MetricsConnection> metrics) {
548    if (query.getConsistency() == Consistency.STRONG) {
549      return requestReplica.apply(RegionReplicaUtil.DEFAULT_REPLICA_ID);
550    }
551    // user specifies a replica id explicitly, just send request to the specific replica
552    if (query.getReplicaId() >= 0) {
553      return requestReplica.apply(query.getReplicaId());
554    }
555    // Timeline consistent read, where we may send requests to other region replicas
556    CompletableFuture<T> primaryFuture = requestReplica.apply(RegionReplicaUtil.DEFAULT_REPLICA_ID);
557    CompletableFuture<T> future = new CompletableFuture<>();
558    connect(primaryFuture, future, Optional.empty());
559    long startNs = System.nanoTime();
560    // after the getRegionLocations, all the locations for the replicas of this region should have
561    // been cached, so it is not big deal to locate them again when actually sending requests to
562    // these replicas.
563    addListener(locator.getRegionLocations(tableName, row, locateType, false, rpcTimeoutNs),
564      (locs, error) -> {
565        if (error != null) {
566          LOG.warn(
567            "Failed to locate all the replicas for table={}, row='{}', locateType={}" +
568              " give up timeline consistent read",
569            tableName, Bytes.toStringBinary(row), locateType, error);
570          return;
571        }
572        if (locs.size() <= 1) {
573          LOG.warn(
574            "There are no secondary replicas for region {}, give up timeline consistent read",
575            locs.getDefaultRegionLocation().getRegion());
576          return;
577        }
578        long delayNs = primaryCallTimeoutNs - (System.nanoTime() - startNs);
579        if (delayNs <= 0) {
580          sendRequestsToSecondaryReplicas(requestReplica, locs, future, metrics);
581        } else {
582          retryTimer.newTimeout(
583            timeout -> sendRequestsToSecondaryReplicas(requestReplica, locs, future, metrics),
584            delayNs, TimeUnit.NANOSECONDS);
585        }
586      });
587    return future;
588  }
589
590  // validate for well-formedness
591  static void validatePut(Put put, int maxKeyValueSize) throws IllegalArgumentException {
592    if (put.isEmpty()) {
593      throw new IllegalArgumentException("No columns to insert");
594    }
595    if (maxKeyValueSize > 0) {
596      for (List<Cell> list : put.getFamilyCellMap().values()) {
597        for (Cell cell : list) {
598          if (cell.getSerializedSize() > maxKeyValueSize) {
599            throw new IllegalArgumentException("KeyValue size too large");
600          }
601        }
602      }
603    }
604  }
605
606  /**
607   * Select the priority for the rpc call.
608   * <p/>
609   * The rules are:
610   * <ol>
611   * <li>If user set a priority explicitly, then just use it.</li>
612   * <li>For system table, use {@link HConstants#SYSTEMTABLE_QOS}.</li>
613   * <li>For other tables, use {@link HConstants#NORMAL_QOS}.</li>
614   * </ol>
615   * @param priority the priority set by user, can be {@link HConstants#PRIORITY_UNSET}.
616   * @param tableName the table we operate on
617   */
618  static int calcPriority(int priority, TableName tableName) {
619    if (priority != HConstants.PRIORITY_UNSET) {
620      return priority;
621    } else {
622      return getPriority(tableName);
623    }
624  }
625
626  static int getPriority(TableName tableName) {
627    if (tableName.isSystemTable()) {
628      return HConstants.SYSTEMTABLE_QOS;
629    } else {
630      return HConstants.NORMAL_QOS;
631    }
632  }
633
634  static <T> CompletableFuture<T> getOrFetch(AtomicReference<T> cacheRef,
635      AtomicReference<CompletableFuture<T>> futureRef, boolean reload,
636      Supplier<CompletableFuture<T>> fetch, Predicate<T> validator, String type) {
637    for (;;) {
638      if (!reload) {
639        T value = cacheRef.get();
640        if (value != null && validator.test(value)) {
641          return CompletableFuture.completedFuture(value);
642        }
643      }
644      LOG.trace("{} cache is null, try fetching from registry", type);
645      if (futureRef.compareAndSet(null, new CompletableFuture<>())) {
646        LOG.debug("Start fetching{} from registry", type);
647        CompletableFuture<T> future = futureRef.get();
648        addListener(fetch.get(), (value, error) -> {
649          if (error != null) {
650            LOG.debug("Failed to fetch {} from registry", type, error);
651            futureRef.getAndSet(null).completeExceptionally(error);
652            return;
653          }
654          LOG.debug("The fetched {} is {}", type, value);
655          // Here we update cache before reset future, so it is possible that someone can get a
656          // stale value. Consider this:
657          // 1. update cacheRef
658          // 2. someone clears the cache and relocates again
659          // 3. the futureRef is not null so the old future is used.
660          // 4. we clear futureRef and complete the future in it with the value being
661          // cleared in step 2.
662          // But we do not think it is a big deal as it rarely happens, and even if it happens, the
663          // caller will retry again later, no correctness problems.
664          cacheRef.set(value);
665          futureRef.set(null);
666          future.complete(value);
667        });
668        return future;
669      } else {
670        CompletableFuture<T> future = futureRef.get();
671        if (future != null) {
672          return future;
673        }
674      }
675    }
676  }
677
678  static void updateStats(Optional<ServerStatisticTracker> optStats,
679      Optional<MetricsConnection> optMetrics, ServerName serverName, MultiResponse resp) {
680    if (!optStats.isPresent() && !optMetrics.isPresent()) {
681      // ServerStatisticTracker and MetricsConnection are both not present, just return
682      return;
683    }
684    resp.getResults().forEach((regionName, regionResult) -> {
685      ClientProtos.RegionLoadStats stat = regionResult.getStat();
686      if (stat == null) {
687        LOG.error("No ClientProtos.RegionLoadStats found for server={}, region={}", serverName,
688          Bytes.toStringBinary(regionName));
689        return;
690      }
691      RegionLoadStats regionLoadStats = ProtobufUtil.createRegionLoadStats(stat);
692      optStats.ifPresent(
693        stats -> ResultStatsUtil.updateStats(stats, serverName, regionName, regionLoadStats));
694      optMetrics.ifPresent(
695        metrics -> ResultStatsUtil.updateStats(metrics, serverName, regionName, regionLoadStats));
696    });
697  }
698}