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 org.apache.hadoop.hbase.client.AsyncRegionLocatorHelper.canUpdateOnError;
021import static org.apache.hadoop.hbase.client.AsyncRegionLocatorHelper.createRegionLocations;
022import static org.apache.hadoop.hbase.client.AsyncRegionLocatorHelper.isGood;
023import static org.apache.hadoop.hbase.client.AsyncRegionLocatorHelper.removeRegionLocation;
024import static org.apache.hadoop.hbase.client.AsyncRegionLocatorHelper.replaceRegionLocation;
025
026import java.util.concurrent.CompletableFuture;
027import java.util.concurrent.atomic.AtomicReference;
028import org.apache.hadoop.hbase.HRegionLocation;
029import org.apache.hadoop.hbase.RegionLocations;
030import org.apache.hadoop.hbase.ServerName;
031import org.apache.yetus.audience.InterfaceAudience;
032
033/**
034 * The asynchronous locator for meta region.
035 */
036@InterfaceAudience.Private
037class AsyncMetaRegionLocator {
038
039  private final ConnectionRegistry registry;
040
041  private final AtomicReference<RegionLocations> metaRegionLocations = new AtomicReference<>();
042
043  private final AtomicReference<CompletableFuture<RegionLocations>> metaRelocateFuture =
044    new AtomicReference<>();
045
046  AsyncMetaRegionLocator(ConnectionRegistry registry) {
047    this.registry = registry;
048  }
049
050  /**
051   * Get the region locations for meta region. If the location for the given replica is not
052   * available in the cached locations, then fetch from the HBase cluster.
053   * <p/>
054   * The <code>replicaId</code> parameter is important. If the region replication config for meta
055   * region is changed, then the cached region locations may not have the locations for new
056   * replicas. If we do not check the location for the given replica, we will always return the
057   * cached region locations and cause an infinite loop.
058   */
059  CompletableFuture<RegionLocations> getRegionLocations(int replicaId, boolean reload) {
060    return ConnectionUtils.getOrFetch(metaRegionLocations, metaRelocateFuture, reload,
061      registry::getMetaRegionLocations, locs -> isGood(locs, replicaId), "meta region location");
062  }
063
064  private HRegionLocation getCacheLocation(HRegionLocation loc) {
065    RegionLocations locs = metaRegionLocations.get();
066    return locs != null ? locs.getRegionLocation(loc.getRegion().getReplicaId()) : null;
067  }
068
069  private void addLocationToCache(HRegionLocation loc) {
070    for (;;) {
071      int replicaId = loc.getRegion().getReplicaId();
072      RegionLocations oldLocs = metaRegionLocations.get();
073      if (oldLocs == null) {
074        RegionLocations newLocs = createRegionLocations(loc);
075        if (metaRegionLocations.compareAndSet(null, newLocs)) {
076          return;
077        }
078      }
079      HRegionLocation oldLoc = oldLocs.getRegionLocation(replicaId);
080      if (oldLoc != null && (oldLoc.getSeqNum() > loc.getSeqNum() ||
081        oldLoc.getServerName().equals(loc.getServerName()))) {
082        return;
083      }
084      RegionLocations newLocs = replaceRegionLocation(oldLocs, loc);
085      if (metaRegionLocations.compareAndSet(oldLocs, newLocs)) {
086        return;
087      }
088    }
089  }
090
091  private void removeLocationFromCache(HRegionLocation loc) {
092    for (;;) {
093      RegionLocations oldLocs = metaRegionLocations.get();
094      if (oldLocs == null) {
095        return;
096      }
097      HRegionLocation oldLoc = oldLocs.getRegionLocation(loc.getRegion().getReplicaId());
098      if (!canUpdateOnError(loc, oldLoc)) {
099        return;
100      }
101      RegionLocations newLocs = removeRegionLocation(oldLocs, loc.getRegion().getReplicaId());
102      if (metaRegionLocations.compareAndSet(oldLocs, newLocs)) {
103        return;
104      }
105    }
106  }
107
108  void updateCachedLocationOnError(HRegionLocation loc, Throwable exception) {
109    AsyncRegionLocatorHelper.updateCachedLocationOnError(loc, exception, this::getCacheLocation,
110      this::addLocationToCache, this::removeLocationFromCache, null);
111  }
112
113  void clearCache() {
114    metaRegionLocations.set(null);
115  }
116
117  void clearCache(ServerName serverName) {
118    for (;;) {
119      RegionLocations locs = metaRegionLocations.get();
120      if (locs == null) {
121        return;
122      }
123      RegionLocations newLocs = locs.removeByServer(serverName);
124      if (locs == newLocs) {
125        return;
126      }
127      if (newLocs.isEmpty()) {
128        newLocs = null;
129      }
130      if (metaRegionLocations.compareAndSet(locs, newLocs)) {
131        return;
132      }
133    }
134  }
135}