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 java.io.IOException;
021import java.io.InterruptedIOException;
022import org.apache.hadoop.hbase.DoNotRetryIOException;
023import org.apache.hadoop.hbase.HBaseIOException;
024import org.apache.hadoop.hbase.HRegionLocation;
025import org.apache.hadoop.hbase.RegionLocations;
026import org.apache.hadoop.hbase.TableName;
027import org.apache.hadoop.hbase.ipc.HBaseRpcController;
028import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
029import org.apache.hadoop.hbase.util.Bytes;
030import org.apache.yetus.audience.InterfaceAudience;
031
032import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
033import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService;
034
035/**
036 * Similar to RegionServerCallable but for the AdminService interface. This service callable assumes
037 * a Table and row and thus does region locating similar to RegionServerCallable. Works against
038 * Admin stub rather than Client stub.
039 */
040@edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "URF_UNREAD_PUBLIC_OR_PROTECTED_FIELD",
041    justification = "stub used by ipc")
042@InterfaceAudience.Private
043public abstract class RegionAdminServiceCallable<T> implements RetryingCallable<T> {
044  protected AdminService.BlockingInterface stub;
045  protected final RpcControllerFactory rpcControllerFactory;
046  private HBaseRpcController controller = null;
047
048  protected final ClusterConnection connection;
049  protected HRegionLocation location;
050  protected final TableName tableName;
051  protected final byte[] row;
052  protected final int replicaId;
053
054  public RegionAdminServiceCallable(ClusterConnection connection,
055    RpcControllerFactory rpcControllerFactory, TableName tableName, byte[] row) {
056    this(connection, rpcControllerFactory, null, tableName, row);
057  }
058
059  public RegionAdminServiceCallable(ClusterConnection connection,
060    RpcControllerFactory rpcControllerFactory, HRegionLocation location, TableName tableName,
061    byte[] row) {
062    this(connection, rpcControllerFactory, location, tableName, row,
063      RegionReplicaUtil.DEFAULT_REPLICA_ID);
064  }
065
066  public RegionAdminServiceCallable(ClusterConnection connection,
067    RpcControllerFactory rpcControllerFactory, HRegionLocation location, TableName tableName,
068    byte[] row, int replicaId) {
069    this.connection = connection;
070    this.rpcControllerFactory = rpcControllerFactory;
071    this.location = location;
072    this.tableName = tableName;
073    this.row = row;
074    this.replicaId = replicaId;
075  }
076
077  @Override
078  public void prepare(boolean reload) throws IOException {
079    if (Thread.interrupted()) {
080      throw new InterruptedIOException();
081    }
082    if (reload || location == null) {
083      location = getLocation(!reload);
084    }
085    if (location == null) {
086      // With this exception, there will be a retry.
087      throw new HBaseIOException(getExceptionMessage());
088    }
089    this.setStub(connection.getAdmin(location.getServerName()));
090  }
091
092  protected void setStub(AdminService.BlockingInterface stub) {
093    this.stub = stub;
094  }
095
096  public HRegionLocation getLocation(boolean useCache) throws IOException {
097    RegionLocations rl = getRegionLocations(connection, tableName, row, useCache, replicaId);
098    if (rl == null) {
099      throw new HBaseIOException(getExceptionMessage());
100    }
101    HRegionLocation location = rl.getRegionLocation(replicaId);
102    if (location == null) {
103      throw new HBaseIOException(getExceptionMessage());
104    }
105
106    return location;
107  }
108
109  @Override
110  public void throwable(Throwable t, boolean retrying) {
111    if (location != null) {
112      connection.updateCachedLocations(tableName, location.getRegionInfo().getRegionName(), row, t,
113        location.getServerName());
114    }
115  }
116
117  /** Returns {@link Connection} instance used by this Callable. */
118  Connection getConnection() {
119    return this.connection;
120  }
121
122  // subclasses can override this.
123  protected String getExceptionMessage() {
124    return "There is no location" + " table=" + tableName + " ,replica=" + replicaId + ", row="
125      + Bytes.toStringBinary(row);
126  }
127
128  @Override
129  public String getExceptionMessageAdditionalDetail() {
130    return null;
131  }
132
133  @Override
134  public long sleep(long pause, int tries) {
135    return ConnectionUtils.getPauseTime(pause, tries);
136  }
137
138  public static RegionLocations getRegionLocations(ClusterConnection connection,
139    TableName tableName, byte[] row, boolean useCache, int replicaId)
140    throws RetriesExhaustedException, DoNotRetryIOException, InterruptedIOException {
141    RegionLocations rl;
142    try {
143      rl = connection.locateRegion(tableName, row, useCache, true, replicaId);
144    } catch (DoNotRetryIOException e) {
145      throw e;
146    } catch (RetriesExhaustedException e) {
147      throw e;
148    } catch (InterruptedIOException e) {
149      throw e;
150    } catch (IOException e) {
151      throw new RetriesExhaustedException("Can't get the location", e);
152    }
153    if (rl == null) {
154      throw new RetriesExhaustedException("Can't get the locations");
155    }
156    return rl;
157  }
158
159  /**
160   * Override that changes Exception from {@link Exception} to {@link IOException}. It also does
161   * setup of an rpcController and calls through to the unimplemented
162   * call(PayloadCarryingRpcController) method; implement this method to add your rpc invocation.
163   */
164  @Override
165  // Same trick as in RegionServerCallable so users don't have to copy/paste so much boilerplate
166  // and so we contain references to protobuf. We can't set priority on the rpcController as
167  // we do in RegionServerCallable because we don't always have a Table when we call.
168  public T call(int callTimeout) throws IOException {
169    this.controller = rpcControllerFactory.newController();
170    this.controller.setPriority(this.tableName);
171    this.controller.setCallTimeout(callTimeout);
172    try {
173      return call(this.controller);
174    } catch (Exception e) {
175      throw ProtobufUtil.handleRemoteException(e);
176    }
177  }
178
179  HBaseRpcController getCurrentPayloadCarryingRpcController() {
180    return this.controller;
181  }
182
183  /**
184   * Run RPC call.
185   * @param rpcController PayloadCarryingRpcController is a mouthful but it at a minimum is a facade
186   *                      on protobuf so we don't have to put protobuf everywhere; we can keep it
187   *                      behind this class.
188   */
189  protected abstract T call(HBaseRpcController rpcController) throws Exception;
190}