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.master.procedure; 019 020import java.io.IOException; 021import java.lang.Thread.UncaughtExceptionHandler; 022import java.util.List; 023import java.util.Set; 024import java.util.concurrent.TimeUnit; 025import org.apache.hadoop.hbase.CallQueueTooBigException; 026import org.apache.hadoop.hbase.DoNotRetryIOException; 027import org.apache.hadoop.hbase.ServerName; 028import org.apache.hadoop.hbase.client.AsyncRegionServerAdmin; 029import org.apache.hadoop.hbase.client.RegionInfo; 030import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException; 031import org.apache.hadoop.hbase.master.MasterServices; 032import org.apache.hadoop.hbase.master.ServerListener; 033import org.apache.hadoop.hbase.master.ServerManager; 034import org.apache.hadoop.hbase.procedure2.ProcedureExecutor; 035import org.apache.hadoop.hbase.procedure2.RemoteProcedureDispatcher; 036import org.apache.hadoop.hbase.regionserver.RegionServerAbortedException; 037import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException; 038import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; 039import org.apache.hadoop.hbase.util.FutureUtils; 040import org.apache.hadoop.ipc.RemoteException; 041import org.apache.yetus.audience.InterfaceAudience; 042import org.slf4j.Logger; 043import org.slf4j.LoggerFactory; 044 045import org.apache.hbase.thirdparty.com.google.common.collect.ArrayListMultimap; 046import org.apache.hbase.thirdparty.com.google.protobuf.ByteString; 047 048import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; 049import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter; 050import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CloseRegionRequest; 051import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ExecuteProceduresRequest; 052import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ExecuteProceduresResponse; 053import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.OpenRegionRequest; 054import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.RemoteProcedureRequest; 055 056/** 057 * A remote procecdure dispatcher for regionservers. 058 */ 059@InterfaceAudience.Private 060public class RSProcedureDispatcher extends RemoteProcedureDispatcher<MasterProcedureEnv, ServerName> 061 implements ServerListener { 062 private static final Logger LOG = LoggerFactory.getLogger(RSProcedureDispatcher.class); 063 064 public static final String RS_RPC_STARTUP_WAIT_TIME_CONF_KEY = 065 "hbase.regionserver.rpc.startup.waittime"; 066 private static final int DEFAULT_RS_RPC_STARTUP_WAIT_TIME = 60000; 067 068 protected final MasterServices master; 069 private final long rsStartupWaitTime; 070 private MasterProcedureEnv procedureEnv; 071 072 public RSProcedureDispatcher(final MasterServices master) { 073 super(master.getConfiguration()); 074 075 this.master = master; 076 this.rsStartupWaitTime = master.getConfiguration().getLong(RS_RPC_STARTUP_WAIT_TIME_CONF_KEY, 077 DEFAULT_RS_RPC_STARTUP_WAIT_TIME); 078 } 079 080 @Override 081 protected UncaughtExceptionHandler getUncaughtExceptionHandler() { 082 return new UncaughtExceptionHandler() { 083 084 @Override 085 public void uncaughtException(Thread t, Throwable e) { 086 LOG.error("Unexpected error caught, this may cause the procedure to hang forever", e); 087 } 088 }; 089 } 090 091 @Override 092 public boolean start() { 093 if (!super.start()) { 094 return false; 095 } 096 setTimeoutExecutorUncaughtExceptionHandler(this::abort); 097 if (master.isStopped()) { 098 LOG.debug("Stopped"); 099 return false; 100 } 101 // Around startup, if failed, some of the below may be set back to null so NPE is possible. 102 ServerManager sm = master.getServerManager(); 103 if (sm == null) { 104 LOG.debug("ServerManager is null"); 105 return false; 106 } 107 sm.registerListener(this); 108 ProcedureExecutor<MasterProcedureEnv> pe = master.getMasterProcedureExecutor(); 109 if (pe == null) { 110 LOG.debug("ProcedureExecutor is null"); 111 return false; 112 } 113 this.procedureEnv = pe.getEnvironment(); 114 if (this.procedureEnv == null) { 115 LOG.debug("ProcedureEnv is null; stopping={}", master.isStopping()); 116 return false; 117 } 118 try { 119 for (ServerName serverName : sm.getOnlineServersList()) { 120 addNode(serverName); 121 } 122 } catch (Exception e) { 123 LOG.info("Failed start", e); 124 return false; 125 } 126 return true; 127 } 128 129 private void abort(Thread t, Throwable e) { 130 LOG.error("Caught error", e); 131 if (!master.isStopped() && !master.isStopping() && !master.isAborted()) { 132 master.abort("Aborting master", e); 133 } 134 } 135 136 @Override 137 public boolean stop() { 138 if (!super.stop()) { 139 return false; 140 } 141 142 master.getServerManager().unregisterListener(this); 143 return true; 144 } 145 146 @Override 147 protected void remoteDispatch(final ServerName serverName, 148 final Set<RemoteProcedure> remoteProcedures) { 149 if (!master.getServerManager().isServerOnline(serverName)) { 150 // fail fast 151 submitTask(new DeadRSRemoteCall(serverName, remoteProcedures)); 152 } else { 153 submitTask(new ExecuteProceduresRemoteCall(serverName, remoteProcedures)); 154 } 155 } 156 157 @Override 158 protected void abortPendingOperations(final ServerName serverName, 159 final Set<RemoteProcedure> operations) { 160 // TODO: Replace with a ServerNotOnlineException() 161 final IOException e = new DoNotRetryIOException("server not online " + serverName); 162 for (RemoteProcedure proc : operations) { 163 proc.remoteCallFailed(procedureEnv, serverName, e); 164 } 165 } 166 167 @Override 168 public void serverAdded(final ServerName serverName) { 169 addNode(serverName); 170 } 171 172 @Override 173 public void serverRemoved(final ServerName serverName) { 174 removeNode(serverName); 175 } 176 177 private interface RemoteProcedureResolver { 178 void dispatchOpenRequests(MasterProcedureEnv env, List<RegionOpenOperation> operations); 179 180 void dispatchCloseRequests(MasterProcedureEnv env, List<RegionCloseOperation> operations); 181 182 void dispatchServerOperations(MasterProcedureEnv env, List<ServerOperation> operations); 183 } 184 185 /** 186 * Fetches {@link org.apache.hadoop.hbase.procedure2.RemoteProcedureDispatcher.RemoteOperation}s 187 * from the given {@code remoteProcedures} and groups them by class of the returned operation. 188 * Then {@code resolver} is used to dispatch {@link RegionOpenOperation}s and 189 * {@link RegionCloseOperation}s. 190 * @param serverName RegionServer to which the remote operations are sent 191 * @param operations Remote procedures which are dispatched to the given server 192 * @param resolver Used to dispatch remote procedures to given server. 193 */ 194 public void splitAndResolveOperation(ServerName serverName, Set<RemoteProcedure> operations, 195 RemoteProcedureResolver resolver) { 196 MasterProcedureEnv env = master.getMasterProcedureExecutor().getEnvironment(); 197 ArrayListMultimap<Class<?>, RemoteOperation> reqsByType = 198 buildAndGroupRequestByType(env, serverName, operations); 199 200 List<RegionOpenOperation> openOps = fetchType(reqsByType, RegionOpenOperation.class); 201 if (!openOps.isEmpty()) { 202 resolver.dispatchOpenRequests(env, openOps); 203 } 204 205 List<RegionCloseOperation> closeOps = fetchType(reqsByType, RegionCloseOperation.class); 206 if (!closeOps.isEmpty()) { 207 resolver.dispatchCloseRequests(env, closeOps); 208 } 209 210 List<ServerOperation> refreshOps = fetchType(reqsByType, ServerOperation.class); 211 if (!refreshOps.isEmpty()) { 212 resolver.dispatchServerOperations(env, refreshOps); 213 } 214 215 if (!reqsByType.isEmpty()) { 216 LOG.warn("unknown request type in the queue: " + reqsByType); 217 } 218 } 219 220 private class DeadRSRemoteCall extends ExecuteProceduresRemoteCall { 221 222 public DeadRSRemoteCall(ServerName serverName, Set<RemoteProcedure> remoteProcedures) { 223 super(serverName, remoteProcedures); 224 } 225 226 @Override 227 public void run() { 228 remoteCallFailed(procedureEnv, 229 new RegionServerStoppedException("Server " + getServerName() + " is not online")); 230 } 231 } 232 233 // ========================================================================== 234 // Compatibility calls 235 // ========================================================================== 236 protected class ExecuteProceduresRemoteCall implements RemoteProcedureResolver, Runnable { 237 238 private final ServerName serverName; 239 240 private final Set<RemoteProcedure> remoteProcedures; 241 242 private int numberOfAttemptsSoFar = 0; 243 private long maxWaitTime = -1; 244 245 private final long rsRpcRetryInterval; 246 private static final String RS_RPC_RETRY_INTERVAL_CONF_KEY = 247 "hbase.regionserver.rpc.retry.interval"; 248 private static final int DEFAULT_RS_RPC_RETRY_INTERVAL = 100; 249 250 private ExecuteProceduresRequest.Builder request = null; 251 252 public ExecuteProceduresRemoteCall(final ServerName serverName, 253 final Set<RemoteProcedure> remoteProcedures) { 254 this.serverName = serverName; 255 this.remoteProcedures = remoteProcedures; 256 this.rsRpcRetryInterval = master.getConfiguration().getLong(RS_RPC_RETRY_INTERVAL_CONF_KEY, 257 DEFAULT_RS_RPC_RETRY_INTERVAL); 258 } 259 260 private AsyncRegionServerAdmin getRsAdmin() throws IOException { 261 return master.getAsyncClusterConnection().getRegionServerAdmin(serverName); 262 } 263 264 protected final ServerName getServerName() { 265 return serverName; 266 } 267 268 private boolean scheduleForRetry(IOException e) { 269 LOG.debug("Request to {} failed, try={}", serverName, numberOfAttemptsSoFar, e); 270 // Should we wait a little before retrying? If the server is starting it's yes. 271 if (e instanceof ServerNotRunningYetException) { 272 long remainingTime = getMaxWaitTime() - EnvironmentEdgeManager.currentTime(); 273 if (remainingTime > 0) { 274 LOG.warn("Waiting a little before retrying {}, try={}, can wait up to {}ms", serverName, 275 numberOfAttemptsSoFar, remainingTime); 276 numberOfAttemptsSoFar++; 277 // Retry every rsRpcRetryInterval millis up to maximum wait time. 278 submitTask(this, rsRpcRetryInterval, TimeUnit.MILLISECONDS); 279 return true; 280 } 281 LOG.warn("{} is throwing ServerNotRunningYetException for {}ms; trying another server", 282 serverName, getMaxWaitTime()); 283 return false; 284 } 285 if (e instanceof DoNotRetryIOException) { 286 LOG.warn("{} tells us DoNotRetry due to {}, try={}, give up", serverName, e.toString(), 287 numberOfAttemptsSoFar); 288 return false; 289 } 290 // This exception is thrown in the rpc framework, where we can make sure that the call has not 291 // been executed yet, so it is safe to mark it as fail. Especially for open a region, we'd 292 // better choose another region server. 293 // Notice that, it is safe to quit only if this is the first time we send request to region 294 // server. Maybe the region server has accepted our request the first time, and then there is 295 // a network error which prevents we receive the response, and the second time we hit a 296 // CallQueueTooBigException, obviously it is not safe to quit here, otherwise it may lead to a 297 // double assign... 298 if (e instanceof CallQueueTooBigException && numberOfAttemptsSoFar == 0) { 299 LOG.warn("request to {} failed due to {}, try={}, this usually because" 300 + " server is overloaded, give up", serverName, e.toString(), numberOfAttemptsSoFar); 301 return false; 302 } 303 // Always retry for other exception types if the region server is not dead yet. 304 if (!master.getServerManager().isServerOnline(serverName)) { 305 LOG.warn("Request to {} failed due to {}, try={} and the server is not online, give up", 306 serverName, e.toString(), numberOfAttemptsSoFar); 307 return false; 308 } 309 if (e instanceof RegionServerAbortedException || e instanceof RegionServerStoppedException) { 310 // A better way is to return true here to let the upper layer quit, and then schedule a 311 // background task to check whether the region server is dead. And if it is dead, call 312 // remoteCallFailed to tell the upper layer. Keep retrying here does not lead to incorrect 313 // result, but waste some resources. 314 LOG.warn("{} is aborted or stopped, for safety we still need to" 315 + " wait until it is fully dead, try={}", serverName, numberOfAttemptsSoFar); 316 } else { 317 LOG.warn("request to {} failed due to {}, try={}, retrying...", serverName, e.toString(), 318 numberOfAttemptsSoFar); 319 } 320 numberOfAttemptsSoFar++; 321 // Add some backoff here as the attempts rise otherwise if a stuck condition, will fill logs 322 // with failed attempts. None of our backoff classes -- RetryCounter or ClientBackoffPolicy 323 // -- fit here nicely so just do something simple; increment by rsRpcRetryInterval millis * 324 // retry^2 on each try 325 // up to max of 10 seconds (don't want to back off too much in case of situation change). 326 submitTask(this, 327 Math.min(rsRpcRetryInterval * (this.numberOfAttemptsSoFar * this.numberOfAttemptsSoFar), 328 10 * 1000), 329 TimeUnit.MILLISECONDS); 330 return true; 331 } 332 333 private long getMaxWaitTime() { 334 if (this.maxWaitTime < 0) { 335 // This is the max attempts, not retries, so it should be at least 1. 336 this.maxWaitTime = EnvironmentEdgeManager.currentTime() + rsStartupWaitTime; 337 } 338 return this.maxWaitTime; 339 } 340 341 private IOException unwrapException(IOException e) { 342 if (e instanceof RemoteException) { 343 e = ((RemoteException) e).unwrapRemoteException(); 344 } 345 return e; 346 } 347 348 @Override 349 public void run() { 350 request = ExecuteProceduresRequest.newBuilder(); 351 if (LOG.isTraceEnabled()) { 352 LOG.trace("Building request with operations count=" + remoteProcedures.size()); 353 } 354 splitAndResolveOperation(getServerName(), remoteProcedures, this); 355 356 try { 357 sendRequest(getServerName(), request.build()); 358 } catch (IOException e) { 359 e = unwrapException(e); 360 // TODO: In the future some operation may want to bail out early. 361 // TODO: How many times should we retry (use numberOfAttemptsSoFar) 362 if (!scheduleForRetry(e)) { 363 remoteCallFailed(procedureEnv, e); 364 } 365 } 366 } 367 368 @Override 369 public void dispatchOpenRequests(final MasterProcedureEnv env, 370 final List<RegionOpenOperation> operations) { 371 request.addOpenRegion(buildOpenRegionRequest(env, getServerName(), operations)); 372 } 373 374 @Override 375 public void dispatchCloseRequests(final MasterProcedureEnv env, 376 final List<RegionCloseOperation> operations) { 377 for (RegionCloseOperation op : operations) { 378 request.addCloseRegion(op.buildCloseRegionRequest(getServerName())); 379 } 380 } 381 382 @Override 383 public void dispatchServerOperations(MasterProcedureEnv env, List<ServerOperation> operations) { 384 operations.stream().map(o -> o.buildRequest()).forEachOrdered(request::addProc); 385 } 386 387 // will be overridden in test. 388 protected ExecuteProceduresResponse sendRequest(final ServerName serverName, 389 final ExecuteProceduresRequest request) throws IOException { 390 return FutureUtils.get(getRsAdmin().executeProcedures(request)); 391 } 392 393 protected final void remoteCallFailed(final MasterProcedureEnv env, final IOException e) { 394 for (RemoteProcedure proc : remoteProcedures) { 395 proc.remoteCallFailed(env, getServerName(), e); 396 } 397 } 398 } 399 400 private static OpenRegionRequest buildOpenRegionRequest(final MasterProcedureEnv env, 401 final ServerName serverName, final List<RegionOpenOperation> operations) { 402 final OpenRegionRequest.Builder builder = OpenRegionRequest.newBuilder(); 403 builder.setServerStartCode(serverName.getStartcode()); 404 builder.setMasterSystemTime(EnvironmentEdgeManager.currentTime()); 405 for (RegionOpenOperation op : operations) { 406 builder.addOpenInfo(op.buildRegionOpenInfoRequest(env)); 407 } 408 return builder.build(); 409 } 410 411 // ========================================================================== 412 // RPC Messages 413 // - ServerOperation: refreshConfig, grant, revoke, ... (TODO) 414 // - RegionOperation: open, close, flush, snapshot, ... 415 // ========================================================================== 416 417 public static final class ServerOperation extends RemoteOperation { 418 419 private final long procId; 420 421 private final Class<?> rsProcClass; 422 423 private final byte[] rsProcData; 424 425 public ServerOperation(RemoteProcedure remoteProcedure, long procId, Class<?> rsProcClass, 426 byte[] rsProcData) { 427 super(remoteProcedure); 428 this.procId = procId; 429 this.rsProcClass = rsProcClass; 430 this.rsProcData = rsProcData; 431 } 432 433 public RemoteProcedureRequest buildRequest() { 434 return RemoteProcedureRequest.newBuilder().setProcId(procId) 435 .setProcClass(rsProcClass.getName()).setProcData(ByteString.copyFrom(rsProcData)).build(); 436 } 437 } 438 439 public static abstract class RegionOperation extends RemoteOperation { 440 protected final RegionInfo regionInfo; 441 protected final long procId; 442 443 protected RegionOperation(RemoteProcedure remoteProcedure, RegionInfo regionInfo, long procId) { 444 super(remoteProcedure); 445 this.regionInfo = regionInfo; 446 this.procId = procId; 447 } 448 } 449 450 public static class RegionOpenOperation extends RegionOperation { 451 452 public RegionOpenOperation(RemoteProcedure remoteProcedure, RegionInfo regionInfo, 453 long procId) { 454 super(remoteProcedure, regionInfo, procId); 455 } 456 457 public OpenRegionRequest.RegionOpenInfo 458 buildRegionOpenInfoRequest(final MasterProcedureEnv env) { 459 return RequestConverter.buildRegionOpenInfo(regionInfo, 460 env.getAssignmentManager().getFavoredNodes(regionInfo), procId); 461 } 462 } 463 464 public static class RegionCloseOperation extends RegionOperation { 465 private final ServerName destinationServer; 466 private boolean evictCache; 467 468 public RegionCloseOperation(RemoteProcedure remoteProcedure, RegionInfo regionInfo, long procId, 469 ServerName destinationServer, boolean evictCache) { 470 super(remoteProcedure, regionInfo, procId); 471 this.destinationServer = destinationServer; 472 this.evictCache = evictCache; 473 } 474 475 public ServerName getDestinationServer() { 476 return destinationServer; 477 } 478 479 public CloseRegionRequest buildCloseRegionRequest(final ServerName serverName) { 480 return ProtobufUtil.buildCloseRegionRequest(serverName, regionInfo.getRegionName(), 481 getDestinationServer(), procId, evictCache); 482 483 } 484 } 485}