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.util;
019
020import java.io.IOException;
021import java.net.ConnectException;
022import java.net.UnknownHostException;
023import java.util.Arrays;
024import java.util.List;
025import java.util.Set;
026import java.util.concurrent.atomic.AtomicBoolean;
027import java.util.concurrent.atomic.AtomicInteger;
028import org.apache.hadoop.hbase.ServerName;
029import org.apache.hadoop.hbase.TableName;
030import org.apache.hadoop.hbase.client.AsyncRegionServerAdmin;
031import org.apache.hadoop.hbase.client.RegionInfo;
032import org.apache.hadoop.hbase.exceptions.ConnectionClosedException;
033import org.apache.hadoop.hbase.master.MasterServices;
034import org.apache.hadoop.hbase.master.procedure.RSProcedureDispatcher;
035import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException;
036import org.slf4j.Logger;
037import org.slf4j.LoggerFactory;
038
039import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
040import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos;
041import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
042
043/**
044 * Test implementation of RSProcedureDispatcher that throws desired errors for testing purpose.
045 */
046public class RSProcDispatcher extends RSProcedureDispatcher {
047
048  private static final Logger LOG = LoggerFactory.getLogger(RSProcDispatcher.class);
049
050  /** Config key for the fail-fast retry limit, shared with the test so the two cannot drift. */
051  static final String FAIL_FAST_LIMIT_KEY = "hbase.master.rs.remote.proc.fail.fast.limit";
052
053  private static final List<IOException> ERRORS =
054    Arrays.asList(new ConnectionClosedException("test connection closed error..."),
055      new UnknownHostException("test unknown host error..."),
056      new ConnectException("test connect error..."));
057
058  private static final AtomicInteger ERROR_IDX = new AtomicInteger();
059
060  // Injection is driven by the test and bound to a target table, not a global call count:
061  // remoteDispatch() fires for every remote procedure in the cluster (startup, table creation,
062  // chores, background assignments), so counting calls drifts and misses the operations under test.
063  private static final AtomicBoolean INJECT = new AtomicBoolean(false);
064  private static final AtomicInteger VICTIMS_REMAINING = new AtomicInteger(0);
065  private static volatile TableName targetTable;
066
067  // Fail-fast retry limit after which the master schedules an SCP; read from conf to match test.
068  private final int failFastLimit;
069
070  /**
071   * Fails the next {@code n} open/close-region requests for {@code table} with connection errors
072   * until the fail-fast retry limit is exhausted, so the master schedules an SCP. Call right before
073   * the operations under test.
074   */
075  static void injectErrorsForNextRequests(TableName table, int n) {
076    ERROR_IDX.set(0);
077    targetTable = table;
078    VICTIMS_REMAINING.set(n);
079    INJECT.set(true);
080  }
081
082  /** Stops error injection. Safe to call unconditionally, e.g. from test teardown. */
083  static void stopInjecting() {
084    INJECT.set(false);
085    VICTIMS_REMAINING.set(0);
086    targetTable = null;
087  }
088
089  public RSProcDispatcher(MasterServices master) {
090    super(master);
091    this.failFastLimit = master.getConfiguration().getInt(FAIL_FAST_LIMIT_KEY, 10);
092  }
093
094  @Override
095  protected void remoteDispatch(final ServerName serverName,
096    final Set<RemoteProcedure> remoteProcedures) {
097    if (!master.getServerManager().isServerOnline(serverName)) {
098      // fail fast
099      submitTask(new DeadRSRemoteCall(serverName, remoteProcedures));
100    } else {
101      submitTask(new TestExecuteProceduresRemoteCall(serverName, remoteProcedures));
102    }
103  }
104
105  /**
106   * True if the request opens or closes a region of the injection target table. Open requests carry
107   * a full RegionInfo; close requests carry a REGION_NAME specifier the table is parsed from.
108   */
109  private static boolean targetsInjectionTable(AdminProtos.ExecuteProceduresRequest request) {
110    TableName table = targetTable;
111    if (table == null) {
112      return false;
113    }
114    for (AdminProtos.OpenRegionRequest open : request.getOpenRegionList()) {
115      for (AdminProtos.OpenRegionRequest.RegionOpenInfo info : open.getOpenInfoList()) {
116        if (table.equals(ProtobufUtil.toTableName(info.getRegion().getTableName()))) {
117          return true;
118        }
119      }
120    }
121    for (AdminProtos.CloseRegionRequest close : request.getCloseRegionList()) {
122      HBaseProtos.RegionSpecifier region = close.getRegion();
123      if (
124        region.getType() == HBaseProtos.RegionSpecifier.RegionSpecifierType.REGION_NAME
125          && table.equals(RegionInfo.getTable(region.getValue().toByteArray()))
126      ) {
127        return true;
128      }
129    }
130    return false;
131  }
132
133  class TestExecuteProceduresRemoteCall extends ExecuteProceduresRemoteCall {
134
135    // attempts: retries of this single request instance (mirrors the dispatcher's
136    // numberOfAttemptsSoFar). injectErrors: whether this instance is failed with injected errors,
137    // decided once on the first call and kept across its retries.
138    private int attempts = 0;
139    private Boolean injectErrors = null;
140
141    public TestExecuteProceduresRemoteCall(ServerName serverName,
142      Set<RemoteProcedure> remoteProcedures) {
143      super(serverName, remoteProcedures);
144    }
145
146    @Override
147    public AdminProtos.ExecuteProceduresResponse sendRequest(final ServerName serverName,
148      final AdminProtos.ExecuteProceduresRequest request) throws IOException {
149      if (injectErrors == null) {
150        // Claim a slot only for a target-table open/close request, once per instance.
151        injectErrors =
152          INJECT.get() && targetsInjectionTable(request) && VICTIMS_REMAINING.getAndDecrement() > 0;
153      }
154      LOG.info("sendRequest() req: {}, attempts: {}, injectErrors: {}", request, attempts,
155        injectErrors);
156      if (!injectErrors) {
157        return FutureUtils.get(getRsAdmin().executeProcedures(request));
158      }
159      // Throw a connection error each attempt until the retry limit is exhausted (-> SCP). On the
160      // last attempt run the real open/close first so the region still recovers.
161      if (attempts++ >= failFastLimit - 1) {
162        FutureUtils.get(getRsAdmin().executeProcedures(request));
163      }
164      throw ERRORS.get(ERROR_IDX.getAndIncrement() % ERRORS.size());
165    }
166
167    private AsyncRegionServerAdmin getRsAdmin() {
168      return master.getAsyncClusterConnection().getRegionServerAdmin(getServerName());
169    }
170  }
171
172  private class DeadRSRemoteCall extends ExecuteProceduresRemoteCall {
173
174    public DeadRSRemoteCall(ServerName serverName, Set<RemoteProcedure> remoteProcedures) {
175      super(serverName, remoteProcedures);
176    }
177
178    @Override
179    public void run() {
180      remoteCallFailed(master.getMasterProcedureExecutor().getEnvironment(),
181        new RegionServerStoppedException("Server " + getServerName() + " is not online"));
182    }
183  }
184}