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.assignment; 019 020import edu.umd.cs.findbugs.annotations.NonNull; 021import java.io.IOException; 022import java.util.ArrayList; 023import java.util.Collection; 024import java.util.Collections; 025import java.util.Comparator; 026import java.util.HashMap; 027import java.util.HashSet; 028import java.util.List; 029import java.util.Map; 030import java.util.Set; 031import java.util.SortedSet; 032import java.util.TreeSet; 033import java.util.concurrent.CompletableFuture; 034import java.util.concurrent.Future; 035import java.util.concurrent.TimeUnit; 036import java.util.concurrent.atomic.AtomicBoolean; 037import java.util.concurrent.locks.Condition; 038import java.util.concurrent.locks.ReentrantLock; 039import java.util.function.Consumer; 040import java.util.function.Function; 041import java.util.stream.Collectors; 042import java.util.stream.Stream; 043import org.apache.hadoop.conf.Configuration; 044import org.apache.hadoop.hbase.CatalogFamilyFormat; 045import org.apache.hadoop.hbase.DoNotRetryIOException; 046import org.apache.hadoop.hbase.HBaseIOException; 047import org.apache.hadoop.hbase.HConstants; 048import org.apache.hadoop.hbase.PleaseHoldException; 049import org.apache.hadoop.hbase.ServerName; 050import org.apache.hadoop.hbase.TableName; 051import org.apache.hadoop.hbase.UnknownRegionException; 052import org.apache.hadoop.hbase.client.DoNotRetryRegionException; 053import org.apache.hadoop.hbase.client.MasterSwitchType; 054import org.apache.hadoop.hbase.client.RegionInfo; 055import org.apache.hadoop.hbase.client.RegionInfoBuilder; 056import org.apache.hadoop.hbase.client.RegionReplicaUtil; 057import org.apache.hadoop.hbase.client.RegionStatesCount; 058import org.apache.hadoop.hbase.client.Result; 059import org.apache.hadoop.hbase.client.ResultScanner; 060import org.apache.hadoop.hbase.client.Scan; 061import org.apache.hadoop.hbase.client.TableDescriptor; 062import org.apache.hadoop.hbase.client.TableState; 063import org.apache.hadoop.hbase.exceptions.UnexpectedStateException; 064import org.apache.hadoop.hbase.favored.FavoredNodesManager; 065import org.apache.hadoop.hbase.favored.FavoredNodesPromoter; 066import org.apache.hadoop.hbase.master.LoadBalancer; 067import org.apache.hadoop.hbase.master.MasterServices; 068import org.apache.hadoop.hbase.master.MetricsAssignmentManager; 069import org.apache.hadoop.hbase.master.RegionPlan; 070import org.apache.hadoop.hbase.master.RegionState; 071import org.apache.hadoop.hbase.master.RegionState.State; 072import org.apache.hadoop.hbase.master.ServerManager; 073import org.apache.hadoop.hbase.master.TableStateManager; 074import org.apache.hadoop.hbase.master.balancer.FavoredStochasticBalancer; 075import org.apache.hadoop.hbase.master.procedure.HBCKServerCrashProcedure; 076import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv; 077import org.apache.hadoop.hbase.master.procedure.MasterProcedureScheduler; 078import org.apache.hadoop.hbase.master.procedure.ProcedureSyncWait; 079import org.apache.hadoop.hbase.master.procedure.ServerCrashProcedure; 080import org.apache.hadoop.hbase.master.procedure.TruncateRegionProcedure; 081import org.apache.hadoop.hbase.master.region.MasterRegion; 082import org.apache.hadoop.hbase.procedure2.Procedure; 083import org.apache.hadoop.hbase.procedure2.ProcedureEvent; 084import org.apache.hadoop.hbase.procedure2.ProcedureExecutor; 085import org.apache.hadoop.hbase.procedure2.ProcedureInMemoryChore; 086import org.apache.hadoop.hbase.procedure2.util.StringUtils; 087import org.apache.hadoop.hbase.regionserver.SequenceId; 088import org.apache.hadoop.hbase.rsgroup.RSGroupBasedLoadBalancer; 089import org.apache.hadoop.hbase.util.Bytes; 090import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; 091import org.apache.hadoop.hbase.util.FutureUtils; 092import org.apache.hadoop.hbase.util.Pair; 093import org.apache.hadoop.hbase.util.Threads; 094import org.apache.hadoop.hbase.util.VersionInfo; 095import org.apache.hadoop.hbase.zookeeper.MetaTableLocator; 096import org.apache.hadoop.hbase.zookeeper.ZKWatcher; 097import org.apache.yetus.audience.InterfaceAudience; 098import org.apache.zookeeper.KeeperException; 099import org.slf4j.Logger; 100import org.slf4j.LoggerFactory; 101 102import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; 103import org.apache.hadoop.hbase.shaded.protobuf.generated.RegionServerStatusProtos.RegionStateTransition; 104import org.apache.hadoop.hbase.shaded.protobuf.generated.RegionServerStatusProtos.RegionStateTransition.TransitionCode; 105import org.apache.hadoop.hbase.shaded.protobuf.generated.RegionServerStatusProtos.ReportRegionStateTransitionRequest; 106import org.apache.hadoop.hbase.shaded.protobuf.generated.RegionServerStatusProtos.ReportRegionStateTransitionResponse; 107 108/** 109 * The AssignmentManager is the coordinator for region assign/unassign operations. 110 * <ul> 111 * <li>In-memory states of regions and servers are stored in {@link RegionStates}.</li> 112 * <li>hbase:meta state updates are handled by {@link RegionStateStore}.</li> 113 * </ul> 114 * Regions are created by CreateTable, Split, Merge. Regions are deleted by DeleteTable, Split, 115 * Merge. Assigns are triggered by CreateTable, EnableTable, Split, Merge, ServerCrash. Unassigns 116 * are triggered by DisableTable, Split, Merge 117 */ 118@InterfaceAudience.Private 119public class AssignmentManager { 120 private static final Logger LOG = LoggerFactory.getLogger(AssignmentManager.class); 121 122 // TODO: AMv2 123 // - handle region migration from hbase1 to hbase2. 124 // - handle sys table assignment first (e.g. acl, namespace) 125 // - handle table priorities 126 // - If ServerBusyException trying to update hbase:meta, we abort the Master 127 // See updateRegionLocation in RegionStateStore. 128 // 129 // See also 130 // https://docs.google.com/document/d/1eVKa7FHdeoJ1-9o8yZcOTAQbv0u0bblBlCCzVSIn69g/edit#heading=h.ystjyrkbtoq5 131 // for other TODOs. 132 133 public static final String BOOTSTRAP_THREAD_POOL_SIZE_CONF_KEY = 134 "hbase.assignment.bootstrap.thread.pool.size"; 135 136 public static final String ASSIGN_DISPATCH_WAIT_MSEC_CONF_KEY = 137 "hbase.assignment.dispatch.wait.msec"; 138 private static final int DEFAULT_ASSIGN_DISPATCH_WAIT_MSEC = 150; 139 140 public static final String ASSIGN_DISPATCH_WAITQ_MAX_CONF_KEY = 141 "hbase.assignment.dispatch.wait.queue.max.size"; 142 private static final int DEFAULT_ASSIGN_DISPATCH_WAITQ_MAX = 100; 143 144 public static final String RIT_CHORE_INTERVAL_MSEC_CONF_KEY = 145 "hbase.assignment.rit.chore.interval.msec"; 146 private static final int DEFAULT_RIT_CHORE_INTERVAL_MSEC = 60 * 1000; 147 148 public static final String DEAD_REGION_METRIC_CHORE_INTERVAL_MSEC_CONF_KEY = 149 "hbase.assignment.dead.region.metric.chore.interval.msec"; 150 private static final int DEFAULT_DEAD_REGION_METRIC_CHORE_INTERVAL_MSEC = 120 * 1000; 151 152 public static final String ASSIGN_MAX_ATTEMPTS = "hbase.assignment.maximum.attempts"; 153 private static final int DEFAULT_ASSIGN_MAX_ATTEMPTS = Integer.MAX_VALUE; 154 155 public static final String ASSIGN_RETRY_IMMEDIATELY_MAX_ATTEMPTS = 156 "hbase.assignment.retry.immediately.maximum.attempts"; 157 private static final int DEFAULT_ASSIGN_RETRY_IMMEDIATELY_MAX_ATTEMPTS = 3; 158 159 /** Region in Transition metrics threshold time */ 160 public static final String METRICS_RIT_STUCK_WARNING_THRESHOLD = 161 "hbase.metrics.rit.stuck.warning.threshold"; 162 private static final int DEFAULT_RIT_STUCK_WARNING_THRESHOLD = 60 * 1000; 163 public static final String UNEXPECTED_STATE_REGION = "Unexpected state for "; 164 165 public static final String FORCE_REGION_RETAINMENT = "hbase.master.scp.retain.assignment.force"; 166 167 public static final boolean DEFAULT_FORCE_REGION_RETAINMENT = false; 168 169 /** The wait time in millis before checking again if the region's previous RS is back online */ 170 public static final String FORCE_REGION_RETAINMENT_WAIT_INTERVAL = 171 "hbase.master.scp.retain.assignment.force.wait-interval"; 172 173 public static final long DEFAULT_FORCE_REGION_RETAINMENT_WAIT_INTERVAL = 50; 174 175 /** 176 * The number of times to check if the region's previous RS is back online, before giving up and 177 * proceeding with assignment on a new RS 178 */ 179 public static final String FORCE_REGION_RETAINMENT_RETRIES = 180 "hbase.master.scp.retain.assignment.force.retries"; 181 182 public static final int DEFAULT_FORCE_REGION_RETAINMENT_RETRIES = 600; 183 184 private final ProcedureEvent<?> metaAssignEvent = new ProcedureEvent<>("meta assign"); 185 private final ProcedureEvent<?> metaLoadEvent = new ProcedureEvent<>("meta load"); 186 187 private final MetricsAssignmentManager metrics; 188 private final RegionInTransitionChore ritChore; 189 private final DeadServerMetricRegionChore deadMetricChore; 190 private final MasterServices master; 191 192 private final AtomicBoolean running = new AtomicBoolean(false); 193 private final RegionStates regionStates = new RegionStates(); 194 private final RegionStateStore regionStateStore; 195 196 /** 197 * When the operator uses this configuration option, any version between the current cluster 198 * version and the value of "hbase.min.version.move.system.tables" does not trigger any 199 * auto-region movement. Auto-region movement here refers to auto-migration of system table 200 * regions to newer server versions. It is assumed that the configured range of versions does not 201 * require special handling of moving system table regions to higher versioned RegionServer. This 202 * auto-migration is done by {@link #checkIfShouldMoveSystemRegionAsync()}. Example: Let's assume 203 * the cluster is on version 1.4.0 and we have set "hbase.min.version.move.system.tables" as 204 * "2.0.0". Now if we upgrade one RegionServer on 1.4.0 cluster to 1.6.0 (< 2.0.0), then 205 * AssignmentManager will not move hbase:meta, hbase:namespace and other system table regions to 206 * newly brought up RegionServer 1.6.0 as part of auto-migration. However, if we upgrade one 207 * RegionServer on 1.4.0 cluster to 2.2.0 (> 2.0.0), then AssignmentManager will move all system 208 * table regions to newly brought up RegionServer 2.2.0 as part of auto-migration done by 209 * {@link #checkIfShouldMoveSystemRegionAsync()}. "hbase.min.version.move.system.tables" is 210 * introduced as part of HBASE-22923. 211 */ 212 private final String minVersionToMoveSysTables; 213 214 private static final String MIN_VERSION_MOVE_SYS_TABLES_CONFIG = 215 "hbase.min.version.move.system.tables"; 216 private static final String DEFAULT_MIN_VERSION_MOVE_SYS_TABLES_CONFIG = ""; 217 218 private final Map<ServerName, Set<byte[]>> rsReports = new HashMap<>(); 219 220 private final boolean shouldAssignRegionsWithFavoredNodes; 221 private final int assignDispatchWaitQueueMaxSize; 222 private final int assignDispatchWaitMillis; 223 private final int assignMaxAttempts; 224 private final int assignRetryImmediatelyMaxAttempts; 225 226 private final MasterRegion masterRegion; 227 228 private final Object checkIfShouldMoveSystemRegionLock = new Object(); 229 230 private Thread assignThread; 231 232 private final boolean forceRegionRetainment; 233 234 private final long forceRegionRetainmentWaitInterval; 235 236 private final int forceRegionRetainmentRetries; 237 238 private final RegionInTransitionTracker regionInTransitionTracker; 239 240 public AssignmentManager(MasterServices master, MasterRegion masterRegion) { 241 this(master, masterRegion, new RegionStateStore(master, masterRegion)); 242 } 243 244 AssignmentManager(MasterServices master, MasterRegion masterRegion, RegionStateStore stateStore) { 245 this.master = master; 246 this.regionStateStore = stateStore; 247 this.metrics = new MetricsAssignmentManager(); 248 this.regionInTransitionTracker = new RegionInTransitionTracker(metrics::updateRitDuration); 249 this.masterRegion = masterRegion; 250 251 final Configuration conf = master.getConfiguration(); 252 253 // Only read favored nodes if using the favored nodes load balancer. 254 this.shouldAssignRegionsWithFavoredNodes = FavoredStochasticBalancer.class 255 .isAssignableFrom(conf.getClass(HConstants.HBASE_MASTER_LOADBALANCER_CLASS, Object.class)); 256 257 this.assignDispatchWaitMillis = 258 conf.getInt(ASSIGN_DISPATCH_WAIT_MSEC_CONF_KEY, DEFAULT_ASSIGN_DISPATCH_WAIT_MSEC); 259 this.assignDispatchWaitQueueMaxSize = 260 conf.getInt(ASSIGN_DISPATCH_WAITQ_MAX_CONF_KEY, DEFAULT_ASSIGN_DISPATCH_WAITQ_MAX); 261 262 this.assignMaxAttempts = 263 Math.max(1, conf.getInt(ASSIGN_MAX_ATTEMPTS, DEFAULT_ASSIGN_MAX_ATTEMPTS)); 264 this.assignRetryImmediatelyMaxAttempts = conf.getInt(ASSIGN_RETRY_IMMEDIATELY_MAX_ATTEMPTS, 265 DEFAULT_ASSIGN_RETRY_IMMEDIATELY_MAX_ATTEMPTS); 266 267 int ritChoreInterval = 268 conf.getInt(RIT_CHORE_INTERVAL_MSEC_CONF_KEY, DEFAULT_RIT_CHORE_INTERVAL_MSEC); 269 this.ritChore = new RegionInTransitionChore(ritChoreInterval); 270 271 int deadRegionChoreInterval = conf.getInt(DEAD_REGION_METRIC_CHORE_INTERVAL_MSEC_CONF_KEY, 272 DEFAULT_DEAD_REGION_METRIC_CHORE_INTERVAL_MSEC); 273 if (deadRegionChoreInterval > 0) { 274 this.deadMetricChore = new DeadServerMetricRegionChore(deadRegionChoreInterval); 275 } else { 276 this.deadMetricChore = null; 277 } 278 minVersionToMoveSysTables = 279 conf.get(MIN_VERSION_MOVE_SYS_TABLES_CONFIG, DEFAULT_MIN_VERSION_MOVE_SYS_TABLES_CONFIG); 280 281 forceRegionRetainment = 282 conf.getBoolean(FORCE_REGION_RETAINMENT, DEFAULT_FORCE_REGION_RETAINMENT); 283 forceRegionRetainmentWaitInterval = conf.getLong(FORCE_REGION_RETAINMENT_WAIT_INTERVAL, 284 DEFAULT_FORCE_REGION_RETAINMENT_WAIT_INTERVAL); 285 forceRegionRetainmentRetries = 286 conf.getInt(FORCE_REGION_RETAINMENT_RETRIES, DEFAULT_FORCE_REGION_RETAINMENT_RETRIES); 287 } 288 289 private void mirrorMetaLocations() throws IOException, KeeperException { 290 // For compatibility, mirror the meta region state to zookeeper 291 // And we still need to use zookeeper to publish the meta region locations to region 292 // server, so they can serve as ClientMetaService 293 ZKWatcher zk = master.getZooKeeper(); 294 if (zk == null || !zk.getRecoverableZooKeeper().getState().isAlive()) { 295 // this is possible in tests, we do not provide a zk watcher or the zk watcher has been closed 296 return; 297 } 298 Collection<RegionStateNode> metaStates = regionStates.getRegionStateNodes(); 299 for (RegionStateNode metaState : metaStates) { 300 MetaTableLocator.setMetaLocation(zk, metaState.getRegionLocation(), 301 metaState.getRegionInfo().getReplicaId(), metaState.getState()); 302 } 303 int replicaCount = metaStates.size(); 304 // remove extra mirror locations 305 for (String znode : zk.getMetaReplicaNodes()) { 306 int replicaId = zk.getZNodePaths().getMetaReplicaIdFromZNode(znode); 307 if (replicaId >= replicaCount) { 308 MetaTableLocator.deleteMetaLocation(zk, replicaId); 309 } 310 } 311 } 312 313 public void start() throws IOException, KeeperException { 314 if (!running.compareAndSet(false, true)) { 315 return; 316 } 317 318 LOG.trace("Starting assignment manager"); 319 320 // Start the Assignment Thread 321 startAssignmentThread(); 322 // load meta region states. 323 // here we are still in the early steps of active master startup. There is only one thread(us) 324 // can access AssignmentManager and create region node, so here we do not need to lock the 325 // region node. 326 try (ResultScanner scanner = 327 masterRegion.getScanner(new Scan().addFamily(HConstants.CATALOG_FAMILY))) { 328 for (;;) { 329 Result result = scanner.next(); 330 if (result == null) { 331 break; 332 } 333 RegionStateStore 334 .visitMetaEntry((r, regionInfo, state, regionLocation, lastHost, openSeqNum) -> { 335 RegionStateNode regionNode = regionStates.getOrCreateRegionStateNode(regionInfo); 336 regionNode.setState(state); 337 regionNode.setLastHost(lastHost); 338 regionNode.setRegionLocation(regionLocation); 339 regionNode.setOpenSeqNum(openSeqNum); 340 regionInTransitionTracker.handleRegionStateNodeOperation(regionNode); 341 342 if (regionNode.getProcedure() != null) { 343 regionNode.getProcedure().stateLoaded(this, regionNode); 344 } 345 if (regionLocation != null) { 346 // TODO: this could lead to some orphan server state nodes, as it is possible that the 347 // region server is already dead and its SCP has already finished but we have 348 // persisted an opening state on this region server. Finally the TRSP will assign the 349 // region to another region server, so it will not cause critical problems, just waste 350 // some memory as no one will try to cleanup these orphan server state nodes. 351 regionStates.createServer(regionLocation); 352 regionStates.addRegionToServer(regionNode); 353 } 354 if (RegionReplicaUtil.isDefaultReplica(regionInfo.getReplicaId())) { 355 setMetaAssigned(regionInfo, state == State.OPEN); 356 } 357 LOG.debug("Loaded {} {}", TableName.META_TABLE_NAME, regionNode); 358 }, result); 359 } 360 } 361 mirrorMetaLocations(); 362 } 363 364 /** 365 * Create RegionStateNode based on the TRSP list, and attach the TRSP to the RegionStateNode. 366 * <p> 367 * This is used to restore the RIT region list, so we do not need to restore it in the loadingMeta 368 * method below. And it is also very important as now before submitting a TRSP, we need to attach 369 * it to the RegionStateNode, which acts like a guard, so we need to restore this information at 370 * the very beginning, before we start processing any procedures. 371 */ 372 public void setupRIT(List<TransitRegionStateProcedure> procs) { 373 procs.forEach(proc -> { 374 RegionInfo regionInfo = proc.getRegion(); 375 RegionStateNode regionNode = regionStates.getOrCreateRegionStateNode(regionInfo); 376 TransitRegionStateProcedure existingProc = regionNode.getProcedure(); 377 if (existingProc != null) { 378 // This is possible, as we will detach the procedure from the RSN before we 379 // actually finish the procedure. This is because that, we will detach the TRSP from the RSN 380 // during execution, at that time, the procedure has not been marked as done in the pv2 381 // framework yet, so it is possible that we schedule a new TRSP immediately and when 382 // arriving here, we will find out that there are multiple TRSPs for the region. But we can 383 // make sure that, only the last one can take the charge, the previous ones should have all 384 // been finished already. So here we will compare the proc id, the greater one will win. 385 if (existingProc.getProcId() < proc.getProcId()) { 386 // the new one wins, unset and set it to the new one below 387 regionNode.unsetProcedure(existingProc); 388 } else { 389 // the old one wins, skip 390 return; 391 } 392 } 393 LOG.info("Attach {} to {}", proc, regionNode); 394 regionNode.setProcedure(proc); 395 }); 396 } 397 398 public void initializationPostMetaOnline() { 399 // now that we are sure that meta is online, we can set TableStateManger in 400 // regionInTransitionTracker 401 regionInTransitionTracker.setTableStateManager(master.getTableStateManager()); 402 } 403 404 public void stop() { 405 if (!running.compareAndSet(true, false)) { 406 return; 407 } 408 409 LOG.info("Stopping assignment manager"); 410 411 // The AM is started before the procedure executor, 412 // but the actual work will be loaded/submitted only once we have the executor 413 final boolean hasProcExecutor = master.getMasterProcedureExecutor() != null; 414 415 // Remove the RIT chore 416 if (hasProcExecutor) { 417 master.getMasterProcedureExecutor().removeChore(this.ritChore); 418 if (this.deadMetricChore != null) { 419 master.getMasterProcedureExecutor().removeChore(this.deadMetricChore); 420 } 421 } 422 423 // Stop the Assignment Thread 424 stopAssignmentThread(); 425 426 // Stop the RegionStateStore 427 regionStates.clear(); 428 regionInTransitionTracker.stop(); 429 430 // Update meta events (for testing) 431 if (hasProcExecutor) { 432 metaLoadEvent.suspend(); 433 for (RegionInfo hri : getMetaRegionSet()) { 434 setMetaAssigned(hri, false); 435 } 436 } 437 } 438 439 public boolean isRunning() { 440 return running.get(); 441 } 442 443 public Configuration getConfiguration() { 444 return master.getConfiguration(); 445 } 446 447 public MetricsAssignmentManager getAssignmentManagerMetrics() { 448 return metrics; 449 } 450 451 private LoadBalancer getBalancer() { 452 return master.getLoadBalancer(); 453 } 454 455 private FavoredNodesPromoter getFavoredNodePromoter() { 456 return (FavoredNodesPromoter) ((RSGroupBasedLoadBalancer) master.getLoadBalancer()) 457 .getInternalBalancer(); 458 } 459 460 private MasterProcedureEnv getProcedureEnvironment() { 461 return master.getMasterProcedureExecutor().getEnvironment(); 462 } 463 464 private MasterProcedureScheduler getProcedureScheduler() { 465 return getProcedureEnvironment().getProcedureScheduler(); 466 } 467 468 int getAssignMaxAttempts() { 469 return assignMaxAttempts; 470 } 471 472 public boolean isForceRegionRetainment() { 473 return forceRegionRetainment; 474 } 475 476 public long getForceRegionRetainmentWaitInterval() { 477 return forceRegionRetainmentWaitInterval; 478 } 479 480 public int getForceRegionRetainmentRetries() { 481 return forceRegionRetainmentRetries; 482 } 483 484 int getAssignRetryImmediatelyMaxAttempts() { 485 return assignRetryImmediatelyMaxAttempts; 486 } 487 488 public RegionStates getRegionStates() { 489 return regionStates; 490 } 491 492 /** 493 * Returns the regions hosted by the specified server. 494 * <p/> 495 * Notice that, for SCP, after we submit the SCP, no one can change the region list for the 496 * ServerStateNode so we do not need any locks here. And for other usage, this can only give you a 497 * snapshot of the current region list for this server, which means, right after you get the 498 * region list, new regions may be moved to this server or some regions may be moved out from this 499 * server, so you should not use it critically if you need strong consistency. 500 */ 501 public List<RegionInfo> getRegionsOnServer(ServerName serverName) { 502 ServerStateNode serverInfo = regionStates.getServerNode(serverName); 503 if (serverInfo == null) { 504 return Collections.emptyList(); 505 } 506 return serverInfo.getRegionInfoList(); 507 } 508 509 private RegionInfo getRegionInfo(RegionStateNode rsn) { 510 if (rsn.isSplit() && !rsn.getRegionInfo().isSplit()) { 511 // see the comments in markRegionAsSplit on why we need to do this converting. 512 return RegionInfoBuilder.newBuilder(rsn.getRegionInfo()).setSplit(true).setOffline(true) 513 .build(); 514 } else { 515 return rsn.getRegionInfo(); 516 } 517 } 518 519 private Stream<RegionStateNode> getRegionStateNodes(TableName tableName, 520 boolean excludeOfflinedSplitParents) { 521 Stream<RegionStateNode> stream = regionStates.getTableRegionStateNodes(tableName).stream(); 522 if (excludeOfflinedSplitParents) { 523 return stream.filter(rsn -> !rsn.isSplit()); 524 } else { 525 return stream; 526 } 527 } 528 529 public List<RegionInfo> getTableRegions(TableName tableName, 530 boolean excludeOfflinedSplitParents) { 531 return getRegionStateNodes(tableName, excludeOfflinedSplitParents).map(this::getRegionInfo) 532 .collect(Collectors.toList()); 533 } 534 535 public List<Pair<RegionInfo, ServerName>> getTableRegionsAndLocations(TableName tableName, 536 boolean excludeOfflinedSplitParents) { 537 return getRegionStateNodes(tableName, excludeOfflinedSplitParents) 538 .map(rsn -> Pair.newPair(getRegionInfo(rsn), rsn.getRegionLocation())) 539 .collect(Collectors.toList()); 540 } 541 542 public RegionStateStore getRegionStateStore() { 543 return regionStateStore; 544 } 545 546 public List<ServerName> getFavoredNodes(final RegionInfo regionInfo) { 547 return this.shouldAssignRegionsWithFavoredNodes 548 ? getFavoredNodePromoter().getFavoredNodes(regionInfo) 549 : ServerName.EMPTY_SERVER_LIST; 550 } 551 552 // ============================================================================================ 553 // Table State Manager helpers 554 // ============================================================================================ 555 private TableStateManager getTableStateManager() { 556 return master.getTableStateManager(); 557 } 558 559 private boolean isTableEnabled(final TableName tableName) { 560 return getTableStateManager().isTableState(tableName, TableState.State.ENABLED); 561 } 562 563 private boolean isTableDisabled(final TableName tableName) { 564 return getTableStateManager().isTableState(tableName, TableState.State.DISABLED, 565 TableState.State.DISABLING); 566 } 567 568 // ============================================================================================ 569 // META Helpers 570 // ============================================================================================ 571 private boolean isMetaRegion(final RegionInfo regionInfo) { 572 return regionInfo.isMetaRegion(); 573 } 574 575 public boolean isMetaRegion(final byte[] regionName) { 576 return getMetaRegionFromName(regionName) != null; 577 } 578 579 public RegionInfo getMetaRegionFromName(final byte[] regionName) { 580 for (RegionInfo hri : getMetaRegionSet()) { 581 if (Bytes.equals(hri.getRegionName(), regionName)) { 582 return hri; 583 } 584 } 585 return null; 586 } 587 588 public boolean isCarryingMeta(final ServerName serverName) { 589 // TODO: handle multiple meta 590 return isCarryingRegion(serverName, RegionInfoBuilder.FIRST_META_REGIONINFO); 591 } 592 593 private boolean isCarryingRegion(final ServerName serverName, final RegionInfo regionInfo) { 594 // TODO: check for state? 595 final RegionStateNode node = regionStates.getRegionStateNode(regionInfo); 596 return (node != null && serverName.equals(node.getRegionLocation())); 597 } 598 599 private RegionInfo getMetaForRegion(final RegionInfo regionInfo) { 600 // if (regionInfo.isMetaRegion()) return regionInfo; 601 // TODO: handle multiple meta. if the region provided is not meta lookup 602 // which meta the region belongs to. 603 return RegionInfoBuilder.FIRST_META_REGIONINFO; 604 } 605 606 // TODO: handle multiple meta. 607 private static final Set<RegionInfo> META_REGION_SET = 608 Collections.singleton(RegionInfoBuilder.FIRST_META_REGIONINFO); 609 610 public Set<RegionInfo> getMetaRegionSet() { 611 return META_REGION_SET; 612 } 613 614 // ============================================================================================ 615 // META Event(s) helpers 616 // ============================================================================================ 617 /** 618 * Notice that, this only means the meta region is available on a RS, but the AM may still be 619 * loading the region states from meta, so usually you need to check {@link #isMetaLoaded()} first 620 * before checking this method, unless you can make sure that your piece of code can only be 621 * executed after AM builds the region states. 622 * @see #isMetaLoaded() 623 */ 624 public boolean isMetaAssigned() { 625 return metaAssignEvent.isReady(); 626 } 627 628 public boolean isMetaRegionInTransition() { 629 return !isMetaAssigned(); 630 } 631 632 /** 633 * Notice that this event does not mean the AM has already finished region state rebuilding. See 634 * the comment of {@link #isMetaAssigned()} for more details. 635 * @see #isMetaAssigned() 636 */ 637 public boolean waitMetaAssigned(Procedure<?> proc, RegionInfo regionInfo) { 638 return getMetaAssignEvent(getMetaForRegion(regionInfo)).suspendIfNotReady(proc); 639 } 640 641 private void setMetaAssigned(RegionInfo metaRegionInfo, boolean assigned) { 642 assert isMetaRegion(metaRegionInfo) : "unexpected non-meta region " + metaRegionInfo; 643 ProcedureEvent<?> metaAssignEvent = getMetaAssignEvent(metaRegionInfo); 644 if (assigned) { 645 metaAssignEvent.wake(getProcedureScheduler()); 646 } else { 647 metaAssignEvent.suspend(); 648 } 649 } 650 651 private ProcedureEvent<?> getMetaAssignEvent(RegionInfo metaRegionInfo) { 652 assert isMetaRegion(metaRegionInfo) : "unexpected non-meta region " + metaRegionInfo; 653 // TODO: handle multiple meta. 654 return metaAssignEvent; 655 } 656 657 /** 658 * Wait until AM finishes the meta loading, i.e, the region states rebuilding. 659 * @see #isMetaLoaded() 660 * @see #waitMetaAssigned(Procedure, RegionInfo) 661 */ 662 public boolean waitMetaLoaded(Procedure<?> proc) { 663 return metaLoadEvent.suspendIfNotReady(proc); 664 } 665 666 /** 667 * This method will be called in master initialization method after calling 668 * {@link #processOfflineRegions()}, as in processOfflineRegions we will generate assign 669 * procedures for offline regions, which may be conflict with creating table. 670 * <p/> 671 * This is a bit dirty, should be reconsidered after we decide whether to keep the 672 * {@link #processOfflineRegions()} method. 673 */ 674 public void wakeMetaLoadedEvent() { 675 metaLoadEvent.wake(getProcedureScheduler()); 676 assert isMetaLoaded() : "expected meta to be loaded"; 677 } 678 679 /** 680 * Return whether AM finishes the meta loading, i.e, the region states rebuilding. 681 * @see #isMetaAssigned() 682 * @see #waitMetaLoaded(Procedure) 683 */ 684 public boolean isMetaLoaded() { 685 return metaLoadEvent.isReady(); 686 } 687 688 /** 689 * Start a new thread to check if there are region servers whose versions are higher than others. 690 * If so, move all system table regions to RS with the highest version to keep compatibility. The 691 * reason is, RS in new version may not be able to access RS in old version when there are some 692 * incompatible changes. 693 * <p> 694 * This method is called when a new RegionServer is added to cluster only. 695 * </p> 696 */ 697 public void checkIfShouldMoveSystemRegionAsync() { 698 // TODO: Fix this thread. If a server is killed and a new one started, this thread thinks that 699 // it should 'move' the system tables from the old server to the new server but 700 // ServerCrashProcedure is on it; and it will take care of the assign without dataloss. 701 if (this.master.getServerManager().countOfRegionServers() <= 1) { 702 return; 703 } 704 // This thread used to run whenever there was a change in the cluster. The ZooKeeper 705 // childrenChanged notification came in before the nodeDeleted message and so this method 706 // cold run before a ServerCrashProcedure could run. That meant that this thread could see 707 // a Crashed Server before ServerCrashProcedure and it could find system regions on the 708 // crashed server and go move them before ServerCrashProcedure had a chance; could be 709 // dataloss too if WALs were not recovered. 710 new Thread(() -> { 711 try { 712 synchronized (checkIfShouldMoveSystemRegionLock) { 713 List<RegionPlan> plans = new ArrayList<>(); 714 // TODO: I don't think this code does a good job if all servers in cluster have same 715 // version. It looks like it will schedule unnecessary moves. 716 for (ServerName server : getExcludedServersForSystemTable()) { 717 if (master.getServerManager().isServerDead(server)) { 718 // TODO: See HBASE-18494 and HBASE-18495. Though getExcludedServersForSystemTable() 719 // considers only online servers, the server could be queued for dead server 720 // processing. As region assignments for crashed server is handled by 721 // ServerCrashProcedure, do NOT handle them here. The goal is to handle this through 722 // regular flow of LoadBalancer as a favored node and not to have this special 723 // handling. 724 continue; 725 } 726 List<RegionInfo> regionsShouldMove = getSystemTables(server); 727 if (!regionsShouldMove.isEmpty()) { 728 for (RegionInfo regionInfo : regionsShouldMove) { 729 // null value for dest forces destination server to be selected by balancer 730 RegionPlan plan = new RegionPlan(regionInfo, server, null); 731 if (regionInfo.isMetaRegion()) { 732 // Must move meta region first. 733 LOG.info("Async MOVE of {} to newer Server={}", regionInfo.getEncodedName(), 734 server); 735 moveAsync(plan); 736 } else { 737 plans.add(plan); 738 } 739 } 740 } 741 for (RegionPlan plan : plans) { 742 LOG.info("Async MOVE of {} to newer Server={}", plan.getRegionInfo().getEncodedName(), 743 server); 744 moveAsync(plan); 745 } 746 } 747 } 748 } catch (Throwable t) { 749 LOG.error(t.toString(), t); 750 } 751 }).start(); 752 } 753 754 private List<RegionInfo> getSystemTables(ServerName serverName) { 755 ServerStateNode serverNode = regionStates.getServerNode(serverName); 756 if (serverNode == null) { 757 return Collections.emptyList(); 758 } 759 return serverNode.getSystemRegionInfoList(); 760 } 761 762 private void preTransitCheck(RegionStateNode regionNode, RegionState.State[] expectedStates) 763 throws HBaseIOException { 764 if (regionNode.getProcedure() != null) { 765 throw new HBaseIOException( 766 regionNode + " is currently in transition; pid=" + regionNode.getProcedure().getProcId()); 767 } 768 if (!regionNode.isInState(expectedStates)) { 769 throw new DoNotRetryRegionException(UNEXPECTED_STATE_REGION + regionNode); 770 } 771 if (isTableDisabled(regionNode.getTable())) { 772 throw new DoNotRetryIOException(regionNode.getTable() + " is disabled for " + regionNode); 773 } 774 } 775 776 /** 777 * Create an assign TransitRegionStateProcedure. Makes sure of RegionState. Throws exception if 778 * not appropriate UNLESS override is set. Used by hbck2 but also by straightline 779 * {@link #assign(RegionInfo, ServerName)} and {@link #assignAsync(RegionInfo, ServerName)}. 780 * @see #createAssignProcedure(RegionStateNode, ServerName) for a version that does NO checking 781 * used when only when no checking needed. 782 * @param override If false, check RegionState is appropriate for assign; if not throw exception. 783 */ 784 private TransitRegionStateProcedure createAssignProcedure(RegionInfo regionInfo, ServerName sn, 785 boolean override, boolean force) throws IOException { 786 RegionStateNode regionNode = regionStates.getOrCreateRegionStateNode(regionInfo); 787 regionNode.lock(); 788 try { 789 if (regionNode.isSplit()) { 790 throw new DoNotRetryRegionException(regionNode.getRegionInfo().getEncodedName() 791 + " is a split parent and cannot be assigned"); 792 } 793 if (override) { 794 if (!force) { 795 preTransitCheck(regionNode, STATES_EXPECTED_ON_ASSIGN); 796 } 797 if (regionNode.getProcedure() != null) { 798 regionNode.unsetProcedure(regionNode.getProcedure()); 799 } 800 } else { 801 preTransitCheck(regionNode, STATES_EXPECTED_ON_ASSIGN); 802 } 803 assert regionNode.getProcedure() == null; 804 TransitRegionStateProcedure proc = regionNode.setProcedure( 805 TransitRegionStateProcedure.assign(getProcedureEnvironment(), regionInfo, sn)); 806 regionInTransitionTracker.handleRegionStateNodeOperation(regionNode); 807 return proc; 808 } finally { 809 regionNode.unlock(); 810 } 811 } 812 813 /** 814 * Create an assign TransitRegionStateProcedure. Does NO checking of RegionState. Presumes 815 * appriopriate state ripe for assign. 816 * @see #createAssignProcedure(RegionInfo, ServerName, boolean, boolean) 817 */ 818 private TransitRegionStateProcedure createAssignProcedure(RegionStateNode regionNode, 819 ServerName targetServer) { 820 regionNode.lock(); 821 try { 822 TransitRegionStateProcedure proc = regionNode.setProcedure(TransitRegionStateProcedure 823 .assign(getProcedureEnvironment(), regionNode.getRegionInfo(), targetServer)); 824 regionInTransitionTracker.handleRegionStateNodeOperation(regionNode); 825 return proc; 826 } finally { 827 regionNode.unlock(); 828 } 829 } 830 831 public long assign(RegionInfo regionInfo, ServerName sn) throws IOException { 832 TransitRegionStateProcedure proc = createAssignProcedure(regionInfo, sn, false, false); 833 ProcedureSyncWait.submitAndWaitProcedure(master.getMasterProcedureExecutor(), proc); 834 return proc.getProcId(); 835 } 836 837 public long assign(RegionInfo regionInfo) throws IOException { 838 return assign(regionInfo, null); 839 } 840 841 /** 842 * Submits a procedure that assigns a region to a target server without waiting for it to finish 843 * @param regionInfo the region we would like to assign 844 * @param sn target server name 845 */ 846 public Future<byte[]> assignAsync(RegionInfo regionInfo, ServerName sn) throws IOException { 847 return ProcedureSyncWait.submitProcedure(master.getMasterProcedureExecutor(), 848 createAssignProcedure(regionInfo, sn, false, false)); 849 } 850 851 /** 852 * Submits a procedure that assigns a region without waiting for it to finish 853 * @param regionInfo the region we would like to assign 854 */ 855 public Future<byte[]> assignAsync(RegionInfo regionInfo) throws IOException { 856 return assignAsync(regionInfo, null); 857 } 858 859 public long unassign(RegionInfo regionInfo) throws IOException { 860 RegionStateNode regionNode = regionStates.getRegionStateNode(regionInfo); 861 if (regionNode == null) { 862 throw new UnknownRegionException("No RegionState found for " + regionInfo.getEncodedName()); 863 } 864 TransitRegionStateProcedure proc; 865 regionNode.lock(); 866 try { 867 preTransitCheck(regionNode, STATES_EXPECTED_ON_UNASSIGN_OR_MOVE); 868 proc = TransitRegionStateProcedure.unassign(getProcedureEnvironment(), regionInfo); 869 regionNode.setProcedure(proc); 870 } finally { 871 regionNode.unlock(); 872 } 873 ProcedureSyncWait.submitAndWaitProcedure(master.getMasterProcedureExecutor(), proc); 874 return proc.getProcId(); 875 } 876 877 public TransitRegionStateProcedure createMoveRegionProcedure(RegionInfo regionInfo, 878 ServerName targetServer) throws HBaseIOException { 879 RegionStateNode regionNode = this.regionStates.getRegionStateNode(regionInfo); 880 if (regionNode == null) { 881 throw new UnknownRegionException( 882 "No RegionStateNode found for " + regionInfo.getEncodedName() + "(Closed/Deleted?)"); 883 } 884 TransitRegionStateProcedure proc; 885 regionNode.lock(); 886 try { 887 preTransitCheck(regionNode, STATES_EXPECTED_ON_UNASSIGN_OR_MOVE); 888 regionNode.checkOnline(); 889 proc = TransitRegionStateProcedure.move(getProcedureEnvironment(), regionInfo, targetServer); 890 regionNode.setProcedure(proc); 891 } finally { 892 regionNode.unlock(); 893 } 894 return proc; 895 } 896 897 public void move(RegionInfo regionInfo) throws IOException { 898 TransitRegionStateProcedure proc = createMoveRegionProcedure(regionInfo, null); 899 ProcedureSyncWait.submitAndWaitProcedure(master.getMasterProcedureExecutor(), proc); 900 } 901 902 public Future<byte[]> moveAsync(RegionPlan regionPlan) throws HBaseIOException { 903 TransitRegionStateProcedure proc = 904 createMoveRegionProcedure(regionPlan.getRegionInfo(), regionPlan.getDestination()); 905 return ProcedureSyncWait.submitProcedure(master.getMasterProcedureExecutor(), proc); 906 } 907 908 public Future<byte[]> balance(RegionPlan regionPlan) throws HBaseIOException { 909 ServerName current = 910 this.getRegionStates().getRegionAssignments().get(regionPlan.getRegionInfo()); 911 if (current == null || !current.equals(regionPlan.getSource())) { 912 LOG.debug("Skip region plan {}, source server not match, current region location is {}", 913 regionPlan, current == null ? "(null)" : current); 914 return null; 915 } 916 return moveAsync(regionPlan); 917 } 918 919 // ============================================================================================ 920 // RegionTransition procedures helpers 921 // ============================================================================================ 922 923 /** 924 * Create round-robin assigns. Use on table creation to distribute out regions across cluster. 925 * @return AssignProcedures made out of the passed in <code>hris</code> and a call to the balancer 926 * to populate the assigns with targets chosen using round-robin (default balancer 927 * scheme). If at assign-time, the target chosen is no longer up, thats fine, the 928 * AssignProcedure will ask the balancer for a new target, and so on. 929 */ 930 public TransitRegionStateProcedure[] createRoundRobinAssignProcedures(List<RegionInfo> hris, 931 List<ServerName> serversToExclude) { 932 if (hris.isEmpty()) { 933 return new TransitRegionStateProcedure[0]; 934 } 935 936 if ( 937 serversToExclude != null && this.master.getServerManager().getOnlineServersList().size() == 1 938 ) { 939 LOG.debug("Only one region server found and hence going ahead with the assignment"); 940 serversToExclude = null; 941 } 942 try { 943 // Ask the balancer to assign our regions. Pass the regions en masse. The balancer can do 944 // a better job if it has all the assignments in the one lump. 945 Map<ServerName, List<RegionInfo>> assignments = getBalancer().roundRobinAssignment(hris, 946 this.master.getServerManager().createDestinationServersList(serversToExclude)); 947 // Return mid-method! 948 return createAssignProcedures(assignments); 949 } catch (IOException hioe) { 950 LOG.warn("Failed roundRobinAssignment", hioe); 951 } 952 // If an error above, fall-through to this simpler assign. Last resort. 953 return createAssignProcedures(hris); 954 } 955 956 /** 957 * Create round-robin assigns. Use on table creation to distribute out regions across cluster. 958 * @return AssignProcedures made out of the passed in <code>hris</code> and a call to the balancer 959 * to populate the assigns with targets chosen using round-robin (default balancer 960 * scheme). If at assign-time, the target chosen is no longer up, thats fine, the 961 * AssignProcedure will ask the balancer for a new target, and so on. 962 */ 963 public TransitRegionStateProcedure[] createRoundRobinAssignProcedures(List<RegionInfo> hris) { 964 return createRoundRobinAssignProcedures(hris, null); 965 } 966 967 static int compare(TransitRegionStateProcedure left, TransitRegionStateProcedure right) { 968 if (left.getRegion().isMetaRegion()) { 969 if (right.getRegion().isMetaRegion()) { 970 return RegionInfo.COMPARATOR.compare(left.getRegion(), right.getRegion()); 971 } 972 return -1; 973 } else if (right.getRegion().isMetaRegion()) { 974 return +1; 975 } 976 if (left.getRegion().getTable().isSystemTable()) { 977 if (right.getRegion().getTable().isSystemTable()) { 978 return RegionInfo.COMPARATOR.compare(left.getRegion(), right.getRegion()); 979 } 980 return -1; 981 } else if (right.getRegion().getTable().isSystemTable()) { 982 return +1; 983 } 984 return RegionInfo.COMPARATOR.compare(left.getRegion(), right.getRegion()); 985 } 986 987 /** 988 * Create one TransitRegionStateProcedure to assign a region w/o specifying a target server. This 989 * method is called from HBCK2. 990 * @return an assign or null 991 */ 992 public TransitRegionStateProcedure createOneAssignProcedure(RegionInfo ri, boolean override, 993 boolean force) { 994 TransitRegionStateProcedure trsp = null; 995 try { 996 trsp = createAssignProcedure(ri, null, override, force); 997 } catch (IOException ioe) { 998 LOG.info( 999 "Failed {} assign, override={}" 1000 + (override ? "" : "; set override to by-pass state checks."), 1001 ri.getEncodedName(), override, ioe); 1002 } 1003 return trsp; 1004 } 1005 1006 /** 1007 * Create one TransitRegionStateProcedure to unassign a region. This method is called from HBCK2. 1008 * @return an unassign or null 1009 */ 1010 public TransitRegionStateProcedure createOneUnassignProcedure(RegionInfo ri, boolean override, 1011 boolean force) { 1012 RegionStateNode regionNode = regionStates.getOrCreateRegionStateNode(ri); 1013 TransitRegionStateProcedure trsp = null; 1014 regionNode.lock(); 1015 try { 1016 if (override) { 1017 if (!force) { 1018 preTransitCheck(regionNode, STATES_EXPECTED_ON_UNASSIGN_OR_MOVE); 1019 } 1020 if (regionNode.getProcedure() != null) { 1021 regionNode.unsetProcedure(regionNode.getProcedure()); 1022 } 1023 } else { 1024 // This is where we could throw an exception; i.e. override is false. 1025 preTransitCheck(regionNode, STATES_EXPECTED_ON_UNASSIGN_OR_MOVE); 1026 } 1027 assert regionNode.getProcedure() == null; 1028 trsp = 1029 TransitRegionStateProcedure.unassign(getProcedureEnvironment(), regionNode.getRegionInfo()); 1030 regionNode.setProcedure(trsp); 1031 } catch (IOException ioe) { 1032 // 'override' must be false here. 1033 LOG.info("Failed {} unassign, override=false; set override to by-pass state checks.", 1034 ri.getEncodedName(), ioe); 1035 } finally { 1036 regionNode.unlock(); 1037 } 1038 return trsp; 1039 } 1040 1041 /** 1042 * Create an array of TransitRegionStateProcedure w/o specifying a target server. Used as fallback 1043 * of caller is unable to do {@link #createAssignProcedures(Map)}. 1044 * <p/> 1045 * If no target server, at assign time, we will try to use the former location of the region if 1046 * one exists. This is how we 'retain' the old location across a server restart. 1047 * <p/> 1048 * Should only be called when you can make sure that no one can touch these regions other than 1049 * you. For example, when you are creating or enabling table. Presumes all Regions are in 1050 * appropriate state ripe for assign; no checking of Region state is done in here. 1051 * @see #createAssignProcedures(Map) 1052 */ 1053 public TransitRegionStateProcedure[] createAssignProcedures(List<RegionInfo> hris) { 1054 return hris.stream().map(hri -> regionStates.getOrCreateRegionStateNode(hri)) 1055 .map(regionNode -> createAssignProcedure(regionNode, null)).sorted(AssignmentManager::compare) 1056 .toArray(TransitRegionStateProcedure[]::new); 1057 } 1058 1059 /** 1060 * Tied to {@link #createAssignProcedures(List)} in that it is called if caller is unable to run 1061 * this method. Presumes all Regions are in appropriate state ripe for assign; no checking of 1062 * Region state is done in here. 1063 * @param assignments Map of assignments from which we produce an array of AssignProcedures. 1064 * @return Assignments made from the passed in <code>assignments</code> 1065 * @see #createAssignProcedures(List) 1066 */ 1067 private TransitRegionStateProcedure[] 1068 createAssignProcedures(Map<ServerName, List<RegionInfo>> assignments) { 1069 return assignments.entrySet().stream() 1070 .flatMap(e -> e.getValue().stream().map(hri -> regionStates.getOrCreateRegionStateNode(hri)) 1071 .map(regionNode -> createAssignProcedure(regionNode, e.getKey()))) 1072 .sorted(AssignmentManager::compare).toArray(TransitRegionStateProcedure[]::new); 1073 } 1074 1075 // for creating unassign TRSP when disabling a table or closing excess region replicas 1076 private TransitRegionStateProcedure forceCreateUnssignProcedure(RegionStateNode regionNode) { 1077 regionNode.lock(); 1078 try { 1079 if (regionNode.isInState(State.OFFLINE, State.CLOSED, State.SPLIT)) { 1080 return null; 1081 } 1082 // in general, a split parent should be in CLOSED or SPLIT state, but anyway, let's check it 1083 // here for safety 1084 if (regionNode.getRegionInfo().isSplit()) { 1085 LOG.warn("{} is a split parent but not in CLOSED or SPLIT state", regionNode); 1086 return null; 1087 } 1088 // As in DisableTableProcedure or ModifyTableProcedure, we will hold the xlock for table, so 1089 // we can make sure that this procedure has not been executed yet, as TRSP will hold the 1090 // shared lock for table all the time. So here we will unset it and when it is actually 1091 // executed, it will find that the attach procedure is not itself and quit immediately. 1092 if (regionNode.getProcedure() != null) { 1093 regionNode.unsetProcedure(regionNode.getProcedure()); 1094 } 1095 return regionNode.setProcedure(TransitRegionStateProcedure.unassign(getProcedureEnvironment(), 1096 regionNode.getRegionInfo())); 1097 } finally { 1098 regionNode.unlock(); 1099 } 1100 } 1101 1102 /** 1103 * Called by DisableTableProcedure to unassign all the regions for a table. 1104 */ 1105 public TransitRegionStateProcedure[] createUnassignProceduresForDisabling(TableName tableName) { 1106 return regionStates.getTableRegionStateNodes(tableName).stream() 1107 .map(this::forceCreateUnssignProcedure).filter(p -> p != null) 1108 .toArray(TransitRegionStateProcedure[]::new); 1109 } 1110 1111 private int submitUnassignProcedure(TableName tableName, 1112 Function<RegionStateNode, Boolean> shouldSubmit, Consumer<RegionStateNode> logRIT, 1113 Consumer<TransitRegionStateProcedure> submit) { 1114 int inTransitionCount = 0; 1115 for (RegionStateNode regionNode : regionStates.getTableRegionStateNodes(tableName)) { 1116 regionNode.lock(); 1117 try { 1118 if (shouldSubmit.apply(regionNode)) { 1119 if (regionNode.isTransitionScheduled()) { 1120 logRIT.accept(regionNode); 1121 inTransitionCount++; 1122 continue; 1123 } 1124 if (regionNode.isInState(State.OFFLINE, State.CLOSED, State.SPLIT)) { 1125 continue; 1126 } 1127 submit.accept(regionNode.setProcedure(TransitRegionStateProcedure 1128 .unassign(getProcedureEnvironment(), regionNode.getRegionInfo()))); 1129 } 1130 } finally { 1131 regionNode.unlock(); 1132 } 1133 } 1134 return inTransitionCount; 1135 } 1136 1137 /** 1138 * Called by DisableTableProcedure to unassign all regions for a table. Will skip submit unassign 1139 * procedure if the region is in transition, so you may need to call this method multiple times. 1140 * @param tableName the table for closing excess region replicas 1141 * @param submit for submitting procedure 1142 * @return the number of regions in transition that we can not schedule unassign procedures 1143 */ 1144 public int submitUnassignProcedureForDisablingTable(TableName tableName, 1145 Consumer<TransitRegionStateProcedure> submit) { 1146 return submitUnassignProcedure(tableName, rn -> true, 1147 rn -> LOG.debug("skip scheduling unassign procedure for {} when closing table regions " 1148 + "for disabling since it is in transition", rn), 1149 submit); 1150 } 1151 1152 /** 1153 * Called by ModifyTableProcedure to unassign all the excess region replicas for a table. Will 1154 * skip submit unassign procedure if the region is in transition, so you may need to call this 1155 * method multiple times. 1156 * @param tableName the table for closing excess region replicas 1157 * @param newReplicaCount the new replica count, should be less than current replica count 1158 * @param submit for submitting procedure 1159 * @return the number of regions in transition that we can not schedule unassign procedures 1160 */ 1161 public int submitUnassignProcedureForClosingExcessRegionReplicas(TableName tableName, 1162 int newReplicaCount, Consumer<TransitRegionStateProcedure> submit) { 1163 return submitUnassignProcedure(tableName, 1164 rn -> rn.getRegionInfo().getReplicaId() >= newReplicaCount, 1165 rn -> LOG.debug("skip scheduling unassign procedure for {} when closing excess region " 1166 + "replicas since it is in transition", rn), 1167 submit); 1168 } 1169 1170 private int numberOfUnclosedRegions(TableName tableName, 1171 Function<RegionStateNode, Boolean> shouldSubmit) { 1172 int unclosed = 0; 1173 for (RegionStateNode regionNode : regionStates.getTableRegionStateNodes(tableName)) { 1174 regionNode.lock(); 1175 try { 1176 if (shouldSubmit.apply(regionNode)) { 1177 if (!regionNode.isInState(State.OFFLINE, State.CLOSED, State.SPLIT)) { 1178 unclosed++; 1179 } 1180 } 1181 } finally { 1182 regionNode.unlock(); 1183 } 1184 } 1185 return unclosed; 1186 } 1187 1188 public int numberOfUnclosedRegionsForDisabling(TableName tableName) { 1189 return numberOfUnclosedRegions(tableName, rn -> true); 1190 } 1191 1192 public int numberOfUnclosedExcessRegionReplicas(TableName tableName, int newReplicaCount) { 1193 return numberOfUnclosedRegions(tableName, 1194 rn -> rn.getRegionInfo().getReplicaId() >= newReplicaCount); 1195 } 1196 1197 public SplitTableRegionProcedure createSplitProcedure(final RegionInfo regionToSplit, 1198 final byte[] splitKey) throws IOException { 1199 return new SplitTableRegionProcedure(getProcedureEnvironment(), regionToSplit, splitKey); 1200 } 1201 1202 public TruncateRegionProcedure createTruncateRegionProcedure(final RegionInfo regionToTruncate) 1203 throws IOException { 1204 return new TruncateRegionProcedure(getProcedureEnvironment(), regionToTruncate); 1205 } 1206 1207 public MergeTableRegionsProcedure createMergeProcedure(RegionInfo... ris) throws IOException { 1208 return new MergeTableRegionsProcedure(getProcedureEnvironment(), ris, false); 1209 } 1210 1211 /** 1212 * Delete the region states. This is called by "DeleteTable" 1213 */ 1214 public void deleteTable(final TableName tableName) throws IOException { 1215 final ArrayList<RegionInfo> regions = regionStates.getTableRegionsInfo(tableName); 1216 regionStateStore.deleteRegions(regions); 1217 for (int i = 0; i < regions.size(); ++i) { 1218 final RegionInfo regionInfo = regions.get(i); 1219 regionStates.deleteRegion(regionInfo); 1220 } 1221 } 1222 1223 // ============================================================================================ 1224 // RS Region Transition Report helpers 1225 // ============================================================================================ 1226 private void reportRegionStateTransition(ReportRegionStateTransitionResponse.Builder builder, 1227 ServerStateNode serverNode, List<RegionStateTransition> transitionList) throws IOException { 1228 for (RegionStateTransition transition : transitionList) { 1229 switch (transition.getTransitionCode()) { 1230 case OPENED: 1231 case FAILED_OPEN: 1232 case CLOSED: 1233 assert transition.getRegionInfoCount() == 1 : transition; 1234 final RegionInfo hri = ProtobufUtil.toRegionInfo(transition.getRegionInfo(0)); 1235 long procId = 1236 transition.getProcIdCount() > 0 ? transition.getProcId(0) : Procedure.NO_PROC_ID; 1237 long seqId = 1238 transition.hasOpenSeqNum() ? transition.getOpenSeqNum() : HConstants.NO_SEQNUM; 1239 // On CLOSE, seed the master's flushed-seqid watermark with the region's final flushed 1240 // seqid (reported by the RS). This complements the OPEN-time seed (HBASE-30335) and 1241 // narrows the graceful-close window where a WAL split of a crashed source RS could 1242 // write orphaned recovered.edits for already-durable edits. Uses the same monotonic 1243 // merge(Math::max) seed, so it never regresses a higher value. 1244 if (transition.getTransitionCode() == TransitionCode.CLOSED && seqId >= 0) { 1245 master.getServerManager().advanceRegionFlushedSeq(hri, seqId); 1246 } 1247 updateRegionTransition(serverNode, transition.getTransitionCode(), hri, seqId, procId); 1248 break; 1249 case READY_TO_SPLIT: 1250 case SPLIT: 1251 case SPLIT_REVERTED: 1252 assert transition.getRegionInfoCount() == 3 : transition; 1253 final RegionInfo parent = ProtobufUtil.toRegionInfo(transition.getRegionInfo(0)); 1254 final RegionInfo splitA = ProtobufUtil.toRegionInfo(transition.getRegionInfo(1)); 1255 final RegionInfo splitB = ProtobufUtil.toRegionInfo(transition.getRegionInfo(2)); 1256 updateRegionSplitTransition(serverNode, transition.getTransitionCode(), parent, splitA, 1257 splitB); 1258 break; 1259 case READY_TO_MERGE: 1260 case MERGED: 1261 case MERGE_REVERTED: 1262 assert transition.getRegionInfoCount() == 3 : transition; 1263 final RegionInfo merged = ProtobufUtil.toRegionInfo(transition.getRegionInfo(0)); 1264 final RegionInfo mergeA = ProtobufUtil.toRegionInfo(transition.getRegionInfo(1)); 1265 final RegionInfo mergeB = ProtobufUtil.toRegionInfo(transition.getRegionInfo(2)); 1266 updateRegionMergeTransition(serverNode, transition.getTransitionCode(), merged, mergeA, 1267 mergeB); 1268 break; 1269 } 1270 } 1271 } 1272 1273 public ReportRegionStateTransitionResponse reportRegionStateTransition( 1274 final ReportRegionStateTransitionRequest req) throws PleaseHoldException { 1275 ReportRegionStateTransitionResponse.Builder builder = 1276 ReportRegionStateTransitionResponse.newBuilder(); 1277 ServerName serverName = ProtobufUtil.toServerName(req.getServer()); 1278 ServerStateNode serverNode = regionStates.getServerNode(serverName); 1279 if (serverNode == null) { 1280 LOG.warn("No server node for {}", serverName); 1281 builder.setErrorMessage("No server node for " + serverName); 1282 return builder.build(); 1283 } 1284 // here we have to acquire a read lock instead of a simple exclusive lock. This is because that 1285 // we should not block other reportRegionStateTransition call from the same region server. This 1286 // is not only about performance, but also to prevent dead lock. Think of the meta region is 1287 // also on the same region server and you hold the lock which blocks the 1288 // reportRegionStateTransition for meta, and since meta is not online, you will block inside the 1289 // lock protection to wait for meta online... 1290 serverNode.readLock().lock(); 1291 try { 1292 // we only accept reportRegionStateTransition if the region server is online, see the comment 1293 // above in submitServerCrash method and HBASE-21508 for more details. 1294 if (serverNode.isInState(ServerState.ONLINE)) { 1295 try { 1296 reportRegionStateTransition(builder, serverNode, req.getTransitionList()); 1297 } catch (PleaseHoldException e) { 1298 LOG.trace("Failed transition ", e); 1299 throw e; 1300 } catch (UnsupportedOperationException | IOException e) { 1301 // TODO: at the moment we have a single error message and the RS will abort 1302 // if the master says that one of the region transitions failed. 1303 LOG.warn("Failed transition", e); 1304 builder.setErrorMessage("Failed transition " + e.getMessage()); 1305 } 1306 } else { 1307 LOG.warn("The region server {} is already dead, skip reportRegionStateTransition call", 1308 serverName); 1309 builder.setErrorMessage("You are dead"); 1310 } 1311 } finally { 1312 serverNode.readLock().unlock(); 1313 } 1314 1315 return builder.build(); 1316 } 1317 1318 private void updateRegionTransition(ServerStateNode serverNode, TransitionCode state, 1319 RegionInfo regionInfo, long seqId, long procId) throws IOException { 1320 checkMetaLoaded(regionInfo, procId); 1321 1322 RegionStateNode regionNode = regionStates.getRegionStateNode(regionInfo); 1323 if (regionNode == null) { 1324 // the table/region is gone. maybe a delete, split, merge 1325 throw new UnexpectedStateException(String.format( 1326 "Server %s was trying to transition region %s to %s. but Region is not known.", 1327 serverNode.getServerName(), regionInfo, state)); 1328 } 1329 LOG.trace("Update region transition serverName={} region={} regionState={}", 1330 serverNode.getServerName(), regionNode, state); 1331 1332 regionNode.lock(); 1333 try { 1334 if (!reportTransition(regionNode, serverNode, state, seqId, procId)) { 1335 // Don't log WARN if shutting down cluster; during shutdown. Avoid the below messages: 1336 // 2018-08-13 10:45:10,551 WARN ...AssignmentManager: No matching procedure found for 1337 // rit=OPEN, location=ve0538.halxg.cloudera.com,16020,1533493000958, 1338 // table=IntegrationTestBigLinkedList, region=65ab289e2fc1530df65f6c3d7cde7aa5 transition 1339 // to CLOSED 1340 // These happen because on cluster shutdown, we currently let the RegionServers close 1341 // regions. This is the only time that region close is not run by the Master (so cluster 1342 // goes down fast). Consider changing it so Master runs all shutdowns. 1343 if ( 1344 this.master.getServerManager().isClusterShutdown() && state.equals(TransitionCode.CLOSED) 1345 ) { 1346 LOG.info("RegionServer {} {}", state, regionNode.getRegionInfo().getEncodedName()); 1347 } else { 1348 LOG.warn("No matching procedure found for {} transition on {} to {}", 1349 serverNode.getServerName(), regionNode, state); 1350 } 1351 } 1352 } finally { 1353 regionNode.unlock(); 1354 } 1355 } 1356 1357 private boolean reportTransition(RegionStateNode regionNode, ServerStateNode serverNode, 1358 TransitionCode state, long seqId, long procId) throws IOException { 1359 ServerName serverName = serverNode.getServerName(); 1360 TransitRegionStateProcedure proc = regionNode.getProcedure(); 1361 if (proc == null) { 1362 return false; 1363 } 1364 proc.reportTransition(master.getMasterProcedureExecutor().getEnvironment(), regionNode, 1365 serverName, state, seqId, procId); 1366 return true; 1367 } 1368 1369 private void updateRegionSplitTransition(final ServerStateNode serverNode, 1370 final TransitionCode state, final RegionInfo parent, final RegionInfo hriA, 1371 final RegionInfo hriB) throws IOException { 1372 checkMetaLoaded(parent, Procedure.NO_PROC_ID); 1373 1374 if (state != TransitionCode.READY_TO_SPLIT) { 1375 throw new UnexpectedStateException( 1376 "unsupported split regionState=" + state + " for parent region " + parent 1377 + " maybe an old RS (< 2.0) had the operation in progress"); 1378 } 1379 1380 // sanity check on the request 1381 if (!Bytes.equals(hriA.getEndKey(), hriB.getStartKey())) { 1382 throw new UnsupportedOperationException("unsupported split request with bad keys: parent=" 1383 + parent + " hriA=" + hriA + " hriB=" + hriB); 1384 } 1385 1386 if (!master.isSplitOrMergeEnabled(MasterSwitchType.SPLIT)) { 1387 LOG.warn("Split switch is off! skip split of " + parent); 1388 throw new DoNotRetryIOException( 1389 "Split region " + parent.getRegionNameAsString() + " failed due to split switch off"); 1390 } 1391 1392 // Submit the Split procedure 1393 final byte[] splitKey = hriB.getStartKey(); 1394 if (LOG.isDebugEnabled()) { 1395 LOG.debug("Split request from {}, parent={}, splitKey={}", serverNode.getServerName(), parent, 1396 Bytes.toStringBinary(splitKey)); 1397 } 1398 // Processing this report happens asynchronously from other activities which can mutate 1399 // the region state. For example, a split procedure may already be running for this parent. 1400 // A split procedure cannot succeed if the parent region is no longer open, so we can 1401 // ignore it in that case. 1402 // Note that submitting more than one split procedure for a given region is 1403 // harmless -- the split is fenced in the procedure handling -- but it would be noisy in 1404 // the logs. Only one procedure can succeed. The other procedure(s) would abort during 1405 // initialization and report failure with WARN level logging. 1406 RegionState parentState = regionStates.getRegionState(parent); 1407 if (parentState != null && parentState.isOpened()) { 1408 master.getMasterProcedureExecutor().submitProcedure(createSplitProcedure(parent, splitKey)); 1409 } else { 1410 LOG.info("Ignoring split request from {}, parent={} because parent is unknown or not open", 1411 serverNode.getServerName(), parent); 1412 return; 1413 } 1414 1415 // If the RS is < 2.0 throw an exception to abort the operation, we are handling the split 1416 if (master.getServerManager().getVersionNumber(serverNode.getServerName()) < 0x0200000) { 1417 throw new UnsupportedOperationException( 1418 String.format("Split handled by the master: " + "parent=%s hriA=%s hriB=%s", 1419 parent.getShortNameToLog(), hriA, hriB)); 1420 } 1421 } 1422 1423 private void updateRegionMergeTransition(final ServerStateNode serverNode, 1424 final TransitionCode state, final RegionInfo merged, final RegionInfo hriA, 1425 final RegionInfo hriB) throws IOException { 1426 checkMetaLoaded(merged, Procedure.NO_PROC_ID); 1427 1428 if (state != TransitionCode.READY_TO_MERGE) { 1429 throw new UnexpectedStateException( 1430 "Unsupported merge regionState=" + state + " for regionA=" + hriA + " regionB=" + hriB 1431 + " merged=" + merged + " maybe an old RS (< 2.0) had the operation in progress"); 1432 } 1433 1434 if (!master.isSplitOrMergeEnabled(MasterSwitchType.MERGE)) { 1435 LOG.warn("Merge switch is off! skip merge of regionA=" + hriA + " regionB=" + hriB); 1436 throw new DoNotRetryIOException( 1437 "Merge of regionA=" + hriA + " regionB=" + hriB + " failed because merge switch is off"); 1438 } 1439 1440 // Submit the Merge procedure 1441 if (LOG.isDebugEnabled()) { 1442 LOG.debug("Handling merge request from RS=" + merged + ", merged=" + merged); 1443 } 1444 master.getMasterProcedureExecutor().submitProcedure(createMergeProcedure(hriA, hriB)); 1445 1446 // If the RS is < 2.0 throw an exception to abort the operation, we are handling the merge 1447 if (master.getServerManager().getVersionNumber(serverNode.getServerName()) < 0x0200000) { 1448 throw new UnsupportedOperationException( 1449 String.format("Merge not handled yet: regionState=%s merged=%s hriA=%s hriB=%s", state, 1450 merged, hriA, hriB)); 1451 } 1452 } 1453 1454 // ============================================================================================ 1455 // RS Status update (report online regions) helpers 1456 // ============================================================================================ 1457 /** 1458 * The master will call this method when the RS send the regionServerReport(). The report will 1459 * contains the "online regions". This method will check the the online regions against the 1460 * in-memory state of the AM, and we will log a warn message if there is a mismatch. This is 1461 * because that there is no fencing between the reportRegionStateTransition method and 1462 * regionServerReport method, so there could be race and introduce inconsistency here, but 1463 * actually there is no problem. 1464 * <p/> 1465 * Please see HBASE-21421 and HBASE-21463 for more details. 1466 */ 1467 public void reportOnlineRegions(ServerName serverName, Set<byte[]> regionNames) { 1468 if (!isRunning()) { 1469 return; 1470 } 1471 if (LOG.isTraceEnabled()) { 1472 LOG.trace("ReportOnlineRegions {} regionCount={}, metaLoaded={} {}", serverName, 1473 regionNames.size(), isMetaLoaded(), 1474 regionNames.stream().map(Bytes::toStringBinary).collect(Collectors.toList())); 1475 } 1476 1477 ServerStateNode serverNode = regionStates.getServerNode(serverName); 1478 if (serverNode == null) { 1479 LOG.warn("Got a report from server {} where its server node is null", serverName); 1480 return; 1481 } 1482 serverNode.readLock().lock(); 1483 try { 1484 if (!serverNode.isInState(ServerState.ONLINE)) { 1485 LOG.warn("Got a report from a server result in state {}", serverNode); 1486 return; 1487 } 1488 } finally { 1489 serverNode.readLock().unlock(); 1490 } 1491 1492 // Track the regionserver reported online regions in memory. 1493 synchronized (rsReports) { 1494 rsReports.put(serverName, regionNames); 1495 } 1496 1497 if (regionNames.isEmpty()) { 1498 // nothing to do if we don't have regions 1499 LOG.trace("no online region found on {}", serverName); 1500 return; 1501 } 1502 if (!isMetaLoaded()) { 1503 // we are still on startup, skip checking 1504 return; 1505 } 1506 // The Heartbeat tells us of what regions are on the region serve, check the state. 1507 checkOnlineRegionsReport(serverNode, regionNames); 1508 } 1509 1510 /** 1511 * Close <code>regionName</code> on <code>sn</code> silently and immediately without using a 1512 * Procedure or going via hbase:meta. For case where a RegionServer's hosting of a Region is not 1513 * aligned w/ the Master's accounting of Region state. This is for cleaning up an error in 1514 * accounting. 1515 */ 1516 private void closeRegionSilently(ServerName sn, byte[] regionName) { 1517 try { 1518 RegionInfo ri = CatalogFamilyFormat.parseRegionInfoFromRegionName(regionName); 1519 // Pass -1 for timeout. Means do not wait. 1520 ServerManager.closeRegionSilentlyAndWait(this.master.getAsyncClusterConnection(), sn, ri, -1); 1521 } catch (Exception e) { 1522 LOG.error("Failed trying to close {} on {}", Bytes.toStringBinary(regionName), sn, e); 1523 } 1524 } 1525 1526 /** 1527 * Check that what the RegionServer reports aligns with the Master's image. If disagreement, we 1528 * will tell the RegionServer to expediently close a Region we do not think it should have. 1529 */ 1530 private void checkOnlineRegionsReport(ServerStateNode serverNode, Set<byte[]> regionNames) { 1531 ServerName serverName = serverNode.getServerName(); 1532 for (byte[] regionName : regionNames) { 1533 if (!isRunning()) { 1534 return; 1535 } 1536 RegionStateNode regionNode = regionStates.getRegionStateNodeFromName(regionName); 1537 if (regionNode == null) { 1538 String regionNameAsStr = Bytes.toStringBinary(regionName); 1539 LOG.warn("No RegionStateNode for {} but reported as up on {}; closing...", regionNameAsStr, 1540 serverName); 1541 closeRegionSilently(serverNode.getServerName(), regionName); 1542 continue; 1543 } 1544 final long lag = 1000; 1545 // This is just a fallback check designed to identify unexpected data inconsistencies, so we 1546 // use tryLock to attempt to acquire the lock, and if the lock cannot be acquired, we skip the 1547 // check. This will not cause any additional problems and also prevents the regionServerReport 1548 // call from being stuck for too long which may cause deadlock on region assignment. 1549 if (regionNode.tryLock()) { 1550 try { 1551 long diff = EnvironmentEdgeManager.currentTime() - regionNode.getLastUpdate(); 1552 if (regionNode.isInState(State.OPENING, State.OPEN)) { 1553 // This is possible as a region server has just closed a region but the region server 1554 // report is generated before the closing, but arrive after the closing. Make sure 1555 // there 1556 // is some elapsed time so less false alarms. 1557 if (!regionNode.getRegionLocation().equals(serverName) && diff > lag) { 1558 LOG.warn("Reporting {} server does not match {} (time since last " 1559 + "update={}ms); closing...", serverName, regionNode, diff); 1560 closeRegionSilently(serverNode.getServerName(), regionName); 1561 } 1562 } else if (!regionNode.isInState(State.CLOSING, State.SPLITTING)) { 1563 // So, we can get report that a region is CLOSED or SPLIT because a heartbeat 1564 // came in at about same time as a region transition. Make sure there is some 1565 // elapsed time so less false alarms. 1566 if (diff > lag) { 1567 LOG.warn("Reporting {} state does not match {} (time since last update={}ms)", 1568 serverName, regionNode, diff); 1569 } 1570 } 1571 } finally { 1572 regionNode.unlock(); 1573 } 1574 } else { 1575 LOG.warn( 1576 "Unable to acquire lock for regionNode {}. It is likely that another thread is currently holding the lock. To avoid deadlock, skip execution for now.", 1577 regionNode); 1578 } 1579 } 1580 } 1581 1582 // ============================================================================================ 1583 // RIT chore 1584 // ============================================================================================ 1585 private static class RegionInTransitionChore extends ProcedureInMemoryChore<MasterProcedureEnv> { 1586 public RegionInTransitionChore(final int timeoutMsec) { 1587 super(timeoutMsec); 1588 } 1589 1590 @Override 1591 protected void periodicExecute(final MasterProcedureEnv env) { 1592 final AssignmentManager am = env.getAssignmentManager(); 1593 1594 final RegionInTransitionStat ritStat = am.computeRegionInTransitionStat(); 1595 if (ritStat.hasRegionsOverThreshold()) { 1596 for (RegionState hri : ritStat.getRegionOverThreshold()) { 1597 am.handleRegionOverStuckWarningThreshold(hri.getRegion()); 1598 } 1599 } 1600 1601 // update metrics 1602 am.updateRegionsInTransitionMetrics(ritStat); 1603 } 1604 } 1605 1606 private static class DeadServerMetricRegionChore 1607 extends ProcedureInMemoryChore<MasterProcedureEnv> { 1608 public DeadServerMetricRegionChore(final int timeoutMsec) { 1609 super(timeoutMsec); 1610 } 1611 1612 @Override 1613 protected void periodicExecute(final MasterProcedureEnv env) { 1614 final ServerManager sm = env.getMasterServices().getServerManager(); 1615 final AssignmentManager am = env.getAssignmentManager(); 1616 // To minimize inconsistencies we are not going to snapshot live servers in advance in case 1617 // new servers are added; OTOH we don't want to add heavy sync for a consistent view since 1618 // this is for metrics. Instead, we're going to check each regions as we go; to avoid making 1619 // too many checks, we maintain a local lists of server, limiting us to false negatives. If 1620 // we miss some recently-dead server, we'll just see it next time. 1621 Set<ServerName> recentlyLiveServers = new HashSet<>(); 1622 int deadRegions = 0, unknownRegions = 0; 1623 for (RegionStateNode rsn : am.getRegionStates().getRegionStateNodes()) { 1624 if (rsn.getState() != State.OPEN) { 1625 continue; // Opportunistic check, should quickly skip RITs, offline tables, etc. 1626 } 1627 // Do not need to acquire region state lock as this is only for showing metrics. 1628 ServerName sn = rsn.getRegionLocation(); 1629 State state = rsn.getState(); 1630 if (state != State.OPEN) { 1631 continue; // Mostly skipping RITs that are already being take care of. 1632 } 1633 if (sn == null) { 1634 ++unknownRegions; // Opened on null? 1635 continue; 1636 } 1637 if (recentlyLiveServers.contains(sn)) { 1638 continue; 1639 } 1640 ServerManager.ServerLiveState sls = sm.isServerKnownAndOnline(sn); 1641 switch (sls) { 1642 case LIVE: 1643 recentlyLiveServers.add(sn); 1644 break; 1645 case DEAD: 1646 ++deadRegions; 1647 break; 1648 case UNKNOWN: 1649 ++unknownRegions; 1650 break; 1651 default: 1652 throw new AssertionError("Unexpected " + sls); 1653 } 1654 } 1655 if (deadRegions > 0 || unknownRegions > 0) { 1656 LOG.info("Found {} OPEN regions on dead servers and {} OPEN regions on unknown servers", 1657 deadRegions, unknownRegions); 1658 } 1659 1660 am.updateDeadServerRegionMetrics(deadRegions, unknownRegions); 1661 } 1662 } 1663 1664 public RegionInTransitionStat computeRegionInTransitionStat() { 1665 final RegionInTransitionStat rit = new RegionInTransitionStat(getConfiguration()); 1666 rit.update(this); 1667 return rit; 1668 } 1669 1670 public static class RegionInTransitionStat { 1671 private final int ritThreshold; 1672 1673 private HashMap<String, RegionState> ritsOverThreshold = null; 1674 private long statTimestamp; 1675 private long oldestRITTime = 0; 1676 private int totalRITsTwiceThreshold = 0; 1677 private int totalRITs = 0; 1678 1679 public RegionInTransitionStat(final Configuration conf) { 1680 this.ritThreshold = 1681 conf.getInt(METRICS_RIT_STUCK_WARNING_THRESHOLD, DEFAULT_RIT_STUCK_WARNING_THRESHOLD); 1682 } 1683 1684 public int getRITThreshold() { 1685 return ritThreshold; 1686 } 1687 1688 public long getTimestamp() { 1689 return statTimestamp; 1690 } 1691 1692 public int getTotalRITs() { 1693 return totalRITs; 1694 } 1695 1696 public long getOldestRITTime() { 1697 return oldestRITTime; 1698 } 1699 1700 public int getTotalRITsOverThreshold() { 1701 Map<String, RegionState> m = this.ritsOverThreshold; 1702 return m != null ? m.size() : 0; 1703 } 1704 1705 public boolean hasRegionsTwiceOverThreshold() { 1706 return totalRITsTwiceThreshold > 0; 1707 } 1708 1709 public boolean hasRegionsOverThreshold() { 1710 Map<String, RegionState> m = this.ritsOverThreshold; 1711 return m != null && !m.isEmpty(); 1712 } 1713 1714 public Collection<RegionState> getRegionOverThreshold() { 1715 Map<String, RegionState> m = this.ritsOverThreshold; 1716 return m != null ? m.values() : Collections.emptySet(); 1717 } 1718 1719 public boolean isRegionOverThreshold(final RegionInfo regionInfo) { 1720 Map<String, RegionState> m = this.ritsOverThreshold; 1721 return m != null && m.containsKey(regionInfo.getEncodedName()); 1722 } 1723 1724 public boolean isRegionTwiceOverThreshold(final RegionInfo regionInfo) { 1725 Map<String, RegionState> m = this.ritsOverThreshold; 1726 if (m == null) { 1727 return false; 1728 } 1729 final RegionState state = m.get(regionInfo.getEncodedName()); 1730 if (state == null) { 1731 return false; 1732 } 1733 return (statTimestamp - state.getStamp()) > (ritThreshold * 2L); 1734 } 1735 1736 protected void update(final AssignmentManager am) { 1737 this.statTimestamp = EnvironmentEdgeManager.currentTime(); 1738 update(am.getRegionsStateInTransition(), statTimestamp); 1739 1740 if (LOG.isDebugEnabled() && ritsOverThreshold != null && !ritsOverThreshold.isEmpty()) { 1741 LOG.debug("RITs over threshold: {}", 1742 ritsOverThreshold.entrySet().stream() 1743 .map(e -> e.getKey() + ":" + e.getValue().getState().name()) 1744 .collect(Collectors.joining("\n"))); 1745 } 1746 } 1747 1748 private void update(final Collection<RegionState> regions, final long currentTime) { 1749 for (RegionState state : regions) { 1750 totalRITs++; 1751 final long ritStartedMs = state.getStamp(); 1752 if (ritStartedMs == 0) { 1753 // Don't output bogus values to metrics if they accidentally make it here. 1754 LOG.warn("The RIT {} has no start time", state.getRegion()); 1755 continue; 1756 } 1757 final long ritTime = currentTime - ritStartedMs; 1758 if (ritTime > ritThreshold) { 1759 if (ritsOverThreshold == null) { 1760 ritsOverThreshold = new HashMap<String, RegionState>(); 1761 } 1762 ritsOverThreshold.put(state.getRegion().getEncodedName(), state); 1763 totalRITsTwiceThreshold += (ritTime > (ritThreshold * 2L)) ? 1 : 0; 1764 } 1765 if (oldestRITTime < ritTime) { 1766 oldestRITTime = ritTime; 1767 } 1768 } 1769 } 1770 } 1771 1772 private void updateRegionsInTransitionMetrics(final RegionInTransitionStat ritStat) { 1773 metrics.updateRITOldestAge(ritStat.getOldestRITTime()); 1774 metrics.updateRITCount(ritStat.getTotalRITs()); 1775 metrics.updateRITCountOverThreshold(ritStat.getTotalRITsOverThreshold()); 1776 } 1777 1778 private void updateDeadServerRegionMetrics(int deadRegions, int unknownRegions) { 1779 metrics.updateDeadServerOpenRegions(deadRegions); 1780 metrics.updateUnknownServerOpenRegions(unknownRegions); 1781 } 1782 1783 private void handleRegionOverStuckWarningThreshold(final RegionInfo regionInfo) { 1784 final RegionStateNode regionNode = regionStates.getRegionStateNode(regionInfo); 1785 // if (regionNode.isStuck()) { 1786 LOG.warn("STUCK Region-In-Transition {}", regionNode); 1787 } 1788 1789 // ============================================================================================ 1790 // TODO: Master load/bootstrap 1791 // ============================================================================================ 1792 public void joinCluster() throws IOException { 1793 long startTime = System.nanoTime(); 1794 LOG.debug("Joining cluster..."); 1795 1796 // Scan hbase:meta to build list of existing regions, servers, and assignment. 1797 // hbase:meta is online now or will be. Inside loadMeta, we keep trying. Can't make progress 1798 // w/o meta. 1799 loadMeta(); 1800 1801 while (master.getServerManager().countOfRegionServers() < 1) { 1802 LOG.info("Waiting for RegionServers to join; current count={}", 1803 master.getServerManager().countOfRegionServers()); 1804 Threads.sleep(250); 1805 } 1806 LOG.info("Number of RegionServers={}", master.getServerManager().countOfRegionServers()); 1807 1808 // Start the chores 1809 master.getMasterProcedureExecutor().addChore(this.ritChore); 1810 master.getMasterProcedureExecutor().addChore(this.deadMetricChore); 1811 1812 long costMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime); 1813 LOG.info("Joined the cluster in {}", StringUtils.humanTimeDiff(costMs)); 1814 } 1815 1816 /** 1817 * Create assign procedure for offline regions. Just follow the old 1818 * processofflineServersWithOnlineRegions method. Since now we do not need to deal with dead 1819 * server any more, we only deal with the regions in OFFLINE state in this method. And this is a 1820 * bit strange, that for new regions, we will add it in CLOSED state instead of OFFLINE state, and 1821 * usually there will be a procedure to track them. The processofflineServersWithOnlineRegions is 1822 * a legacy from long ago, as things are going really different now, maybe we do not need this 1823 * method any more. Need to revisit later. 1824 */ 1825 // Public so can be run by the Master as part of the startup. Needs hbase:meta to be online. 1826 // Needs to be done after the table state manager has been started. 1827 public void processOfflineRegions() { 1828 TransitRegionStateProcedure[] procs = 1829 regionStates.getRegionStateNodes().stream().filter(rsn -> rsn.isInState(State.OFFLINE)) 1830 .filter(rsn -> isTableEnabled(rsn.getRegionInfo().getTable())).map(rsn -> { 1831 rsn.lock(); 1832 try { 1833 if (rsn.getProcedure() != null) { 1834 return null; 1835 } else { 1836 return rsn.setProcedure(TransitRegionStateProcedure.assign(getProcedureEnvironment(), 1837 rsn.getRegionInfo(), null)); 1838 } 1839 } finally { 1840 rsn.unlock(); 1841 } 1842 }).filter(p -> p != null).toArray(TransitRegionStateProcedure[]::new); 1843 if (procs.length > 0) { 1844 master.getMasterProcedureExecutor().submitProcedures(procs); 1845 } 1846 } 1847 1848 /* 1849 * AM internal RegionStateStore.RegionStateVisitor implementation. To be used when scanning META 1850 * table for region rows, using RegionStateStore utility methods. RegionStateStore methods will 1851 * convert Result into proper RegionInfo instances, but those would still need to be added into 1852 * AssignmentManager.regionStates in-memory cache. RegionMetaLoadingVisitor.visitRegionState 1853 * method provides the logic for adding RegionInfo instances as loaded from latest META scan into 1854 * AssignmentManager.regionStates. 1855 */ 1856 private class RegionMetaLoadingVisitor implements RegionStateStore.RegionStateVisitor { 1857 1858 @Override 1859 public void visitRegionState(Result result, final RegionInfo regionInfo, final State state, 1860 final ServerName regionLocation, final ServerName lastHost, final long openSeqNum) { 1861 if ( 1862 state == null && regionLocation == null && lastHost == null 1863 && openSeqNum == SequenceId.NO_SEQUENCE_ID 1864 ) { 1865 // This is a row with nothing in it. 1866 LOG.warn("Skipping empty row={}", result); 1867 return; 1868 } 1869 State localState = state; 1870 if (localState == null) { 1871 // No region state column data in hbase:meta table! Are I doing a rolling upgrade from 1872 // hbase1 to hbase2? Am I restoring a SNAPSHOT or otherwise adding a region to hbase:meta? 1873 // In any of these cases, state is empty. For now, presume OFFLINE but there are probably 1874 // cases where we need to probe more to be sure this correct; TODO informed by experience. 1875 LOG.info(regionInfo.getEncodedName() + " regionState=null; presuming " + State.OFFLINE); 1876 localState = State.OFFLINE; 1877 } 1878 RegionStateNode regionNode = regionStates.getOrCreateRegionStateNode(regionInfo); 1879 // Do not need to lock on regionNode, as we can make sure that before we finish loading 1880 // meta, all the related procedures can not be executed. The only exception is for meta 1881 // region related operations, but here we do not load the informations for meta region. 1882 regionNode.setState(localState); 1883 regionNode.setLastHost(lastHost); 1884 regionNode.setRegionLocation(regionLocation); 1885 regionNode.setOpenSeqNum(openSeqNum); 1886 1887 // Note: keep consistent with other methods, see region(Opening|Opened|Closing) 1888 // RIT/ServerCrash handling should take care of the transiting regions. 1889 if ( 1890 localState.matches(State.OPEN, State.OPENING, State.CLOSING, State.SPLITTING, State.MERGING) 1891 ) { 1892 assert regionLocation != null : "found null region location for " + regionNode; 1893 // TODO: this could lead to some orphan server state nodes, as it is possible that the 1894 // region server is already dead and its SCP has already finished but we have 1895 // persisted an opening state on this region server. Finally the TRSP will assign the 1896 // region to another region server, so it will not cause critical problems, just waste 1897 // some memory as no one will try to cleanup these orphan server state nodes. 1898 regionStates.createServer(regionLocation); 1899 regionStates.addRegionToServer(regionNode); 1900 } else if (localState == State.OFFLINE || regionInfo.isOffline()) { 1901 regionStates.addToOfflineRegions(regionNode); 1902 } 1903 if (regionNode.getProcedure() != null) { 1904 regionNode.getProcedure().stateLoaded(AssignmentManager.this, regionNode); 1905 } 1906 // add regions to RIT while visiting the meta 1907 regionInTransitionTracker.handleRegionStateNodeOperation(regionNode); 1908 // If region is supposed to be serve traffic (NOT split and merged) and location of region 1909 // belongs to a dead server mark the region crashed 1910 if ( 1911 regionNode.getRegionLocation() != null && !AssignmentManagerUtil.isSplitOrMerged(regionNode) 1912 && master.getServerManager().isServerDead(regionNode.getRegionLocation()) 1913 ) { 1914 long timeOfCrash = master.getServerManager().getDeadServers() 1915 .getDeathTimestamp(regionNode.getRegionLocation()); 1916 if (timeOfCrash != 0) { 1917 regionNode.crashed(timeOfCrash); 1918 } 1919 regionInTransitionTracker.regionCrashed(regionNode, timeOfCrash); 1920 } 1921 } 1922 }; 1923 1924 /** 1925 * Attempt to load {@code regionInfo} from META, adding any results to the 1926 * {@link #regionStateStore} Is NOT aware of replica regions. 1927 * @param regionInfo the region to be loaded from META. 1928 * @throws IOException If some error occurs while querying META or parsing results. 1929 */ 1930 public void populateRegionStatesFromMeta(@NonNull final RegionInfo regionInfo) 1931 throws IOException { 1932 final String regionEncodedName = RegionInfo.DEFAULT_REPLICA_ID == regionInfo.getReplicaId() 1933 ? regionInfo.getEncodedName() 1934 : RegionInfoBuilder.newBuilder(regionInfo).setReplicaId(RegionInfo.DEFAULT_REPLICA_ID).build() 1935 .getEncodedName(); 1936 populateRegionStatesFromMeta(regionEncodedName); 1937 } 1938 1939 /** 1940 * Attempt to load {@code regionEncodedName} from META, adding any results to the 1941 * {@link #regionStateStore} Is NOT aware of replica regions. 1942 * @param regionEncodedName encoded name for the region to be loaded from META. 1943 * @throws IOException If some error occurs while querying META or parsing results. 1944 */ 1945 public void populateRegionStatesFromMeta(@NonNull String regionEncodedName) throws IOException { 1946 final RegionMetaLoadingVisitor visitor = new RegionMetaLoadingVisitor(); 1947 regionStateStore.visitMetaForRegion(regionEncodedName, visitor); 1948 } 1949 1950 private void loadMeta() throws IOException { 1951 // TODO: use a thread pool 1952 regionStateStore.visitMeta(new RegionMetaLoadingVisitor()); 1953 } 1954 1955 /** 1956 * Used to check if the meta loading is done. 1957 * <p/> 1958 * if not we throw PleaseHoldException since we are rebuilding the RegionStates 1959 * @param hri region to check if it is already rebuild 1960 * @param procId the procedure id for this region operation, or NO_PROC_ID if not available 1961 * @throws PleaseHoldException if meta has not been loaded yet 1962 */ 1963 private void checkMetaLoaded(RegionInfo hri, long procId) throws PleaseHoldException { 1964 if (!isRunning()) { 1965 throw new PleaseHoldException("AssignmentManager not running"); 1966 } 1967 1968 // Check if the procedure is for a critical system table 1969 // Critical system tables can proceed even if meta is not loaded yet 1970 // We are currently making procId available only for the code path which can execute during the 1971 // cluster boot up. In the future, if additional code paths execute during cluster boot up, we 1972 // will need to make procId available for all those code paths. 1973 if (procId != Procedure.NO_PROC_ID) { 1974 Procedure<?> proc = master.getMasterProcedureExecutor().getProcedure(procId); 1975 if (proc != null && proc.isCriticalSystemTable()) { 1976 return; 1977 } 1978 } 1979 1980 boolean meta = isMetaRegion(hri); 1981 boolean metaLoaded = isMetaLoaded(); 1982 if (!meta && !metaLoaded) { 1983 throw new PleaseHoldException("Master not fully online; " + TableName.META_TABLE_NAME + "=" 1984 + meta + ", metaLoaded=" + metaLoaded); 1985 } 1986 } 1987 1988 // ============================================================================================ 1989 // TODO: Metrics 1990 // ============================================================================================ 1991 public int getNumRegionsOpened() { 1992 // TODO: Used by TestRegionPlacement.java and assume monotonically increasing value 1993 return 0; 1994 } 1995 1996 /** 1997 * Usually run by the Master in reaction to server crash during normal processing. Can also be 1998 * invoked via external RPC to effect repair; in the latter case, the 'force' flag is set so we 1999 * push through the SCP though context may indicate already-running-SCP (An old SCP may have 2000 * exited abnormally, or damaged cluster may still have references in hbase:meta to 'Unknown 2001 * Servers' -- servers that are not online or in dead servers list, etc.) 2002 * @param force Set if the request came in externally over RPC (via hbck2). Force means run the 2003 * SCP even if it seems as though there might be an outstanding SCP running. 2004 * @return pid of scheduled SCP or {@link Procedure#NO_PROC_ID} if none scheduled. 2005 */ 2006 public long submitServerCrash(ServerName serverName, boolean shouldSplitWal, boolean force) { 2007 // May be an 'Unknown Server' so handle case where serverNode is null. 2008 ServerStateNode serverNode = regionStates.getServerNode(serverName); 2009 // Remove the in-memory rsReports result 2010 synchronized (rsReports) { 2011 rsReports.remove(serverName); 2012 } 2013 if (serverNode == null) { 2014 if (force) { 2015 LOG.info("Force adding ServerCrashProcedure for {} when server node is null", serverName); 2016 } else { 2017 // for normal case, do not schedule SCP if ServerStateNode is null 2018 LOG.warn("Skip adding ServerCrashProcedure for {} because server node is null", serverName); 2019 return Procedure.NO_PROC_ID; 2020 } 2021 } 2022 2023 ProcedureExecutor<MasterProcedureEnv> procExec = this.master.getMasterProcedureExecutor(); 2024 // We hold the write lock here for fencing on reportRegionStateTransition. Once we set the 2025 // server state to CRASHED, we will no longer accept the reportRegionStateTransition call from 2026 // this server. This is used to simplify the implementation for TRSP and SCP, where we can make 2027 // sure that, the region list fetched by SCP will not be changed any more. 2028 if (serverNode != null) { 2029 serverNode.writeLock().lock(); 2030 } 2031 try { 2032 2033 boolean carryingMeta = isCarryingMeta(serverName); 2034 if (serverNode != null && !serverNode.isInState(ServerState.ONLINE)) { 2035 if (force) { 2036 LOG.info("Force adding ServerCrashProcedure for {} (meta={}) when state is not {}", 2037 serverNode, carryingMeta, ServerState.ONLINE); 2038 } else { 2039 LOG.info("Skip adding ServerCrashProcedure for {} (meta={}) when state is not {}", 2040 serverNode, carryingMeta, ServerState.ONLINE); 2041 return Procedure.NO_PROC_ID; 2042 } 2043 } 2044 MasterProcedureEnv mpe = procExec.getEnvironment(); 2045 // If serverNode == null, then 'Unknown Server'. Schedule HBCKSCP instead. 2046 // HBCKSCP scours Master in-memory state AND hbase;meta for references to 2047 // serverName just-in-case. An SCP that is scheduled when the server is 2048 // 'Unknown' probably originated externally with HBCK2 fix-it tool. 2049 ServerState oldState = null; 2050 if (serverNode != null) { 2051 oldState = serverNode.getState(); 2052 serverNode.setState(ServerState.CRASHED); 2053 } 2054 ServerCrashProcedure scp = force 2055 ? new HBCKServerCrashProcedure(mpe, serverName, shouldSplitWal, carryingMeta) 2056 : new ServerCrashProcedure(mpe, serverName, shouldSplitWal, carryingMeta); 2057 long pid = procExec.submitProcedure(scp); 2058 LOG.info("Scheduled ServerCrashProcedure pid={} for {} (carryingMeta={}){}.", pid, serverName, 2059 carryingMeta, 2060 serverNode == null ? "" : " " + serverNode.toString() + ", oldState=" + oldState); 2061 return pid; 2062 } finally { 2063 if (serverNode != null) { 2064 serverNode.writeLock().unlock(); 2065 } 2066 } 2067 } 2068 2069 public void offlineRegion(final RegionInfo regionInfo) { 2070 // TODO used by MasterRpcServices 2071 RegionStateNode node = regionStates.getRegionStateNode(regionInfo); 2072 if (node != null) { 2073 node.offline(); 2074 } 2075 } 2076 2077 public void onlineRegion(final RegionInfo regionInfo, final ServerName serverName) { 2078 // TODO used by TestSplitTransactionOnCluster.java 2079 } 2080 2081 public Map<ServerName, List<RegionInfo>> 2082 getSnapShotOfAssignment(final Collection<RegionInfo> regions) { 2083 return regionStates.getSnapShotOfAssignment(regions); 2084 } 2085 2086 // ============================================================================================ 2087 // TODO: UTILS/HELPERS? 2088 // ============================================================================================ 2089 /** 2090 * Used by the client (via master) to identify if all regions have the schema updates 2091 * @return Pair indicating the status of the alter command (pending/total) 2092 */ 2093 public Pair<Integer, Integer> getReopenStatus(TableName tableName) { 2094 if (isTableDisabled(tableName)) { 2095 return new Pair<Integer, Integer>(0, 0); 2096 } 2097 2098 final List<RegionState> states = regionStates.getTableRegionStates(tableName); 2099 int ritCount = 0; 2100 for (RegionState regionState : states) { 2101 if (!regionState.isOpened() && !regionState.isSplit()) { 2102 ritCount++; 2103 } 2104 } 2105 return new Pair<Integer, Integer>(ritCount, states.size()); 2106 } 2107 2108 // This comparator sorts the RegionStates by time stamp then Region name. 2109 // Comparing by timestamp alone can lead us to discard different RegionStates that happen 2110 // to share a timestamp. 2111 private final static class RegionStateStampComparator implements Comparator<RegionState> { 2112 @Override 2113 public int compare(final RegionState l, final RegionState r) { 2114 int stampCmp = Long.compare(l.getStamp(), r.getStamp()); 2115 return stampCmp != 0 ? stampCmp : RegionInfo.COMPARATOR.compare(l.getRegion(), r.getRegion()); 2116 } 2117 } 2118 2119 public final static RegionStateStampComparator REGION_STATE_STAMP_COMPARATOR = 2120 new RegionStateStampComparator(); 2121 2122 // ============================================================================================ 2123 // TODO: Region State In Transition 2124 // ============================================================================================ 2125 public boolean hasRegionsInTransition() { 2126 return regionInTransitionTracker.hasRegionsInTransition(); 2127 } 2128 2129 public List<RegionStateNode> getRegionsInTransition() { 2130 return regionInTransitionTracker.getRegionsInTransition(); 2131 } 2132 2133 public int getRegionTransitScheduledCount() { 2134 return regionStates.getRegionTransitScheduledCount(); 2135 } 2136 2137 /** 2138 * Get the number of regions in transition. 2139 */ 2140 public int getRegionsInTransitionCount() { 2141 return regionInTransitionTracker.getRegionsInTransitionCount(); 2142 } 2143 2144 public SortedSet<RegionState> getRegionsStateInTransition() { 2145 final SortedSet<RegionState> rit = new TreeSet<RegionState>(REGION_STATE_STAMP_COMPARATOR); 2146 for (RegionStateNode node : getRegionsInTransition()) { 2147 rit.add(node.toRegionState()); 2148 } 2149 return rit; 2150 } 2151 2152 public List<RegionInfo> getAssignedRegions() { 2153 return regionStates.getAssignedRegions(); 2154 } 2155 2156 /** 2157 * Resolve a cached {@link RegionInfo} from the region name as a {@code byte[]}. 2158 */ 2159 public RegionInfo getRegionInfo(final byte[] regionName) { 2160 final RegionStateNode regionState = regionStates.getRegionStateNodeFromName(regionName); 2161 return regionState != null ? regionState.getRegionInfo() : null; 2162 } 2163 2164 /** 2165 * Resolve a cached {@link RegionInfo} from the encoded region name as a {@code String}. 2166 */ 2167 public RegionInfo getRegionInfo(final String encodedRegionName) { 2168 final RegionStateNode regionState = 2169 regionStates.getRegionStateNodeFromEncodedRegionName(encodedRegionName); 2170 return regionState != null ? regionState.getRegionInfo() : null; 2171 } 2172 2173 // ============================================================================================ 2174 // Expected states on region state transition. 2175 // Notice that there is expected states for transiting to OPENING state, this is because SCP. 2176 // See the comments in regionOpening method for more details. 2177 // ============================================================================================ 2178 private static final State[] STATES_EXPECTED_ON_OPEN = { State.OPENING, // Normal case 2179 State.OPEN // Retrying 2180 }; 2181 2182 private static final State[] STATES_EXPECTED_ON_CLOSING = { State.OPEN, // Normal case 2183 State.CLOSING, // Retrying 2184 State.SPLITTING, // Offline the split parent 2185 State.MERGING // Offline the merge parents 2186 }; 2187 2188 private static final State[] STATES_EXPECTED_ON_CLOSED = { State.CLOSING, // Normal case 2189 State.CLOSED // Retrying 2190 }; 2191 2192 // This is for manually scheduled region assign, can add other states later if we find out other 2193 // usages 2194 private static final State[] STATES_EXPECTED_ON_ASSIGN = { State.CLOSED, State.OFFLINE }; 2195 2196 // We only allow unassign or move a region which is in OPEN state. 2197 private static final State[] STATES_EXPECTED_ON_UNASSIGN_OR_MOVE = { State.OPEN }; 2198 2199 // ============================================================================================ 2200 // Region Status update 2201 // Should only be called in TransitRegionStateProcedure(and related procedures), as the locking 2202 // and pre-assumptions are very tricky. 2203 // ============================================================================================ 2204 private CompletableFuture<Void> transitStateAndUpdate(RegionStateNode regionNode, 2205 RegionState.State newState, RegionState.State... expectedStates) { 2206 RegionState.State state = regionNode.getState(); 2207 try { 2208 regionNode.transitionState(newState, expectedStates); 2209 } catch (UnexpectedStateException e) { 2210 return FutureUtils.failedFuture(e); 2211 } 2212 CompletableFuture<Void> future = regionStateStore.updateRegionLocation(regionNode); 2213 FutureUtils.addListener(future, (r, e) -> { 2214 if (e != null) { 2215 // revert 2216 regionNode.setState(state); 2217 } else { 2218 regionInTransitionTracker.handleRegionStateNodeOperation(regionNode); 2219 } 2220 }); 2221 return future; 2222 } 2223 2224 // should be called within the synchronized block of RegionStateNode 2225 CompletableFuture<Void> regionOpening(RegionStateNode regionNode) { 2226 // As in SCP, for performance reason, there is no TRSP attached with this region, we will not 2227 // update the region state, which means that the region could be in any state when we want to 2228 // assign it after a RS crash. So here we do not pass the expectedStates parameter. 2229 return transitStateAndUpdate(regionNode, State.OPENING).thenAccept(r -> { 2230 ServerStateNode serverNode = regionStates.getServerNode(regionNode.getRegionLocation()); 2231 // Here the server node could be null. For example, we want to assign the region to a given 2232 // region server and it crashes, and it is the region server which holds hbase:meta, then the 2233 // above transitStateAndUpdate call will never succeed until we finishes the SCP for it. But 2234 // after the SCP finishes, the server node will be removed, so when we arrive there, the 2235 // server 2236 // node will be null. This is not a big problem if we skip adding it, as later we will fail to 2237 // execute the remote procedure on the region server and then try to assign to another region 2238 // server 2239 if (serverNode != null) { 2240 serverNode.addRegion(regionNode); 2241 } 2242 // update the operation count metrics 2243 metrics.incrementOperationCounter(); 2244 }); 2245 } 2246 2247 // should be called under the RegionStateNode lock 2248 // Called when a region open attempt has failed. The 'giveUp' parameter says whether we will try 2249 // to open the region again. If true, this is a terminal failure with no more retries: we set 2250 // RegionState.State to FAILED_OPEN and persist it to hbase:meta, and only remove the region from 2251 // its old target server's bookkeeping once that persist succeeds. If false, we are about to 2252 // retry the open with a new plan, so RegionState.State is intentionally left untouched (e.g. 2253 // still OPENING), meta is not touched, and we remove the region from its old target server's 2254 // bookkeeping immediately. In both cases the removal only happens if there was an old location. 2255 CompletableFuture<Void> regionFailedOpen(RegionStateNode regionNode, boolean giveUp) { 2256 RegionState.State state = regionNode.getState(); 2257 ServerName regionLocation = regionNode.getRegionLocation(); 2258 if (!giveUp) { 2259 if (regionLocation != null) { 2260 regionStates.removeRegionFromServer(regionLocation, regionNode); 2261 } 2262 return CompletableFuture.completedFuture(null); 2263 } 2264 regionNode.setState(State.FAILED_OPEN); 2265 regionNode.setRegionLocation(null); 2266 CompletableFuture<Void> future = regionStateStore.updateRegionLocation(regionNode); 2267 FutureUtils.addListener(future, (r, e) -> { 2268 if (e == null) { 2269 if (regionLocation != null) { 2270 regionStates.removeRegionFromServer(regionLocation, regionNode); 2271 } 2272 regionInTransitionTracker.handleRegionStateNodeOperation(regionNode); 2273 } else { 2274 // revert 2275 regionNode.setState(state); 2276 regionNode.setRegionLocation(regionLocation); 2277 } 2278 }); 2279 return future; 2280 } 2281 2282 // should be called under the RegionStateNode lock 2283 CompletableFuture<Void> regionClosing(RegionStateNode regionNode) { 2284 return transitStateAndUpdate(regionNode, State.CLOSING, STATES_EXPECTED_ON_CLOSING) 2285 .thenAccept(r -> { 2286 RegionInfo hri = regionNode.getRegionInfo(); 2287 // Set meta has not initialized early. so people trying to create/edit tables will wait 2288 if (isMetaRegion(hri)) { 2289 setMetaAssigned(hri, false); 2290 } 2291 // update the operation count metrics 2292 metrics.incrementOperationCounter(); 2293 }); 2294 } 2295 2296 // for open and close, they will first be persist to the procedure store in 2297 // RegionRemoteProcedureBase. So here we will first change the in memory state as it is considered 2298 // as succeeded if the persistence to procedure store is succeeded, and then when the 2299 // RegionRemoteProcedureBase is woken up, we will persist the RegionStateNode to hbase:meta. 2300 2301 // should be called under the RegionStateNode lock 2302 void regionOpenedWithoutPersistingToMeta(RegionStateNode regionNode) 2303 throws UnexpectedStateException { 2304 regionNode.transitionState(State.OPEN, STATES_EXPECTED_ON_OPEN); 2305 RegionInfo regionInfo = regionNode.getRegionInfo(); 2306 regionStates.addRegionToServer(regionNode); 2307 regionStates.removeFromFailedOpen(regionInfo); 2308 // HBASE-30335: seed the master's flushed sequence cache with openSeqNum so a subsequent 2309 // WAL split (e.g. source RS crashes after drain-move) recognizes already-durable edits 2310 // instead of writing orphaned recovered.edits. 2311 master.getServerManager().advanceRegionFlushedSeq(regionInfo, regionNode.getOpenSeqNum()); 2312 } 2313 2314 // should be called under the RegionStateNode lock 2315 void regionClosedWithoutPersistingToMeta(RegionStateNode regionNode) 2316 throws UnexpectedStateException { 2317 ServerName regionLocation = regionNode.getRegionLocation(); 2318 regionNode.transitionState(State.CLOSED, STATES_EXPECTED_ON_CLOSED); 2319 regionNode.setRegionLocation(null); 2320 if (regionLocation != null) { 2321 regionNode.setLastHost(regionLocation); 2322 regionStates.removeRegionFromServer(regionLocation, regionNode); 2323 } 2324 } 2325 2326 // should be called under the RegionStateNode lock 2327 CompletableFuture<Void> persistToMeta(RegionStateNode regionNode) { 2328 return regionStateStore.updateRegionLocation(regionNode).thenAccept(r -> { 2329 RegionInfo regionInfo = regionNode.getRegionInfo(); 2330 if (isMetaRegion(regionInfo) && regionNode.getState() == State.OPEN) { 2331 // Usually we'd set a table ENABLED at this stage but hbase:meta is ALWAYs enabled, it 2332 // can't be disabled -- so skip the RPC (besides... enabled is managed by TableStateManager 2333 // which is backed by hbase:meta... Avoid setting ENABLED to avoid having to update state 2334 // on table that contains state. 2335 setMetaAssigned(regionInfo, true); 2336 } 2337 regionInTransitionTracker.handleRegionStateNodeOperation(regionNode); 2338 }); 2339 } 2340 2341 // should be called under the RegionStateNode lock 2342 // for SCP 2343 public CompletableFuture<Void> regionClosedAbnormally(RegionStateNode regionNode) { 2344 RegionState.State state = regionNode.getState(); 2345 ServerName regionLocation = regionNode.getRegionLocation(); 2346 regionNode.setState(State.ABNORMALLY_CLOSED); 2347 regionNode.setRegionLocation(null); 2348 CompletableFuture<Void> future = regionStateStore.updateRegionLocation(regionNode); 2349 FutureUtils.addListener(future, (r, e) -> { 2350 if (e == null) { 2351 if (regionLocation != null) { 2352 regionNode.setLastHost(regionLocation); 2353 regionStates.removeRegionFromServer(regionLocation, regionNode); 2354 } 2355 regionInTransitionTracker.handleRegionStateNodeOperation(regionNode); 2356 } else { 2357 // revert 2358 regionNode.setState(state); 2359 regionNode.setRegionLocation(regionLocation); 2360 } 2361 }); 2362 return future; 2363 } 2364 2365 // ============================================================================================ 2366 // The above methods can only be called in TransitRegionStateProcedure(and related procedures) 2367 // ============================================================================================ 2368 2369 // As soon as a server a crashed, region hosting on that are un-available, this method helps to 2370 // track those un-available regions. This method can only be called from ServerCrashProcedure. 2371 public void markRegionsAsCrashed(List<RegionInfo> regionsOnCrashedServer, 2372 ServerCrashProcedure scp) { 2373 ServerName crashedServerName = scp.getServerName(); 2374 assert crashedServerName != null; 2375 for (RegionInfo regionInfo : regionsOnCrashedServer) { 2376 RegionStateNode node = regionStates.getOrCreateRegionStateNode(regionInfo); 2377 if (crashedServerName.equals(node.getRegionLocation())) { 2378 node.crashed(scp.getSubmittedTime()); 2379 regionInTransitionTracker.regionCrashed(node, scp.getSubmittedTime()); 2380 } else { 2381 LOG.warn("Region {} should be on crashed region server {} but is recorded on {}", 2382 regionInfo, crashedServerName, node.getRegionLocation()); 2383 } 2384 } 2385 } 2386 2387 public void markRegionAsSplit(final RegionInfo parent, final ServerName serverName, 2388 final RegionInfo daughterA, final RegionInfo daughterB) throws IOException { 2389 // Update hbase:meta. Parent will be marked offline and split up in hbase:meta. 2390 // The parent stays in regionStates until cleared when removed by CatalogJanitor. 2391 // Update its state in regionStates to it shows as offline and split when read 2392 // later figuring what regions are in a table and what are not: see 2393 // regionStates#getRegionsOfTable 2394 final RegionStateNode node = regionStates.getOrCreateRegionStateNode(parent); 2395 node.setState(State.SPLIT); 2396 final RegionStateNode nodeA = regionStates.getOrCreateRegionStateNode(daughterA); 2397 nodeA.setState(State.SPLITTING_NEW); 2398 final RegionStateNode nodeB = regionStates.getOrCreateRegionStateNode(daughterB); 2399 nodeB.setState(State.SPLITTING_NEW); 2400 2401 TableDescriptor td = master.getTableDescriptors().get(parent.getTable()); 2402 // TODO: here we just update the parent region info in meta, to set split and offline to true, 2403 // without changing the one in the region node. This is a bit confusing but the region info 2404 // field in RegionStateNode is not expected to be changed in the current design. Need to find a 2405 // possible way to address this problem, or at least adding more comments about the trick to 2406 // deal with this problem, that when you want to filter out split parent, you need to check both 2407 // the RegionState on whether it is split, and also the region info. If one of them matches then 2408 // it is a split parent. And usually only one of them can match, as after restart, the region 2409 // state will be changed from SPLIT to CLOSED. 2410 regionStateStore.splitRegion(parent, daughterA, daughterB, serverName, td); 2411 regionInTransitionTracker.handleRegionStateNodeOperation(node); 2412 regionInTransitionTracker.handleRegionStateNodeOperation(nodeA); 2413 regionInTransitionTracker.handleRegionStateNodeOperation(nodeB); 2414 if (shouldAssignFavoredNodes(parent)) { 2415 List<ServerName> onlineServers = this.master.getServerManager().getOnlineServersList(); 2416 getFavoredNodePromoter().generateFavoredNodesForDaughter(onlineServers, parent, daughterA, 2417 daughterB); 2418 } 2419 } 2420 2421 /** 2422 * When called here, the merge has happened. The merged regions have been unassigned and the above 2423 * markRegionClosed has been called on each so they have been disassociated from a hosting Server. 2424 * The merged region will be open after this call. The merged regions are removed from hbase:meta 2425 * below. Later they are deleted from the filesystem by the catalog janitor running against 2426 * hbase:meta. It notices when the merged region no longer holds references to the old regions 2427 * (References are deleted after a compaction rewrites what the Reference points at but not until 2428 * the archiver chore runs, are the References removed). 2429 */ 2430 public void markRegionAsMerged(final RegionInfo child, final ServerName serverName, 2431 RegionInfo[] mergeParents) throws IOException { 2432 final RegionStateNode node = regionStates.getOrCreateRegionStateNode(child); 2433 for (RegionInfo ri : mergeParents) { 2434 regionStates.deleteRegion(ri); 2435 regionInTransitionTracker.handleRegionDelete(ri); 2436 } 2437 2438 TableDescriptor td = master.getTableDescriptors().get(child.getTable()); 2439 regionStateStore.mergeRegions(child, mergeParents, serverName, td); 2440 regionInTransitionTracker.handleRegionStateNodeOperation(node); 2441 if (shouldAssignFavoredNodes(child)) { 2442 getFavoredNodePromoter().generateFavoredNodesForMergedRegion(child, mergeParents); 2443 } 2444 } 2445 2446 /* 2447 * Favored nodes should be applied only when FavoredNodes balancer is configured and the region 2448 * belongs to a non-system table. 2449 */ 2450 private boolean shouldAssignFavoredNodes(RegionInfo region) { 2451 return this.shouldAssignRegionsWithFavoredNodes 2452 && FavoredNodesManager.isFavoredNodeApplicable(region); 2453 } 2454 2455 // ============================================================================================ 2456 // Assign Queue (Assign/Balance) 2457 // ============================================================================================ 2458 private final ArrayList<RegionStateNode> pendingAssignQueue = new ArrayList<RegionStateNode>(); 2459 private final ReentrantLock assignQueueLock = new ReentrantLock(); 2460 private final Condition assignQueueFullCond = assignQueueLock.newCondition(); 2461 2462 /** 2463 * Add the assign operation to the assignment queue. The pending assignment operation will be 2464 * processed, and each region will be assigned by a server using the balancer. 2465 */ 2466 protected void queueAssign(final RegionStateNode regionNode) { 2467 regionNode.getProcedureEvent().suspend(); 2468 2469 // TODO: quick-start for meta and the other sys-tables? 2470 assignQueueLock.lock(); 2471 try { 2472 pendingAssignQueue.add(regionNode); 2473 if ( 2474 regionNode.isSystemTable() || pendingAssignQueue.size() == 1 2475 || pendingAssignQueue.size() >= assignDispatchWaitQueueMaxSize 2476 ) { 2477 assignQueueFullCond.signal(); 2478 } 2479 } finally { 2480 assignQueueLock.unlock(); 2481 } 2482 } 2483 2484 private void startAssignmentThread() { 2485 assignThread = new Thread(master.getServerName().toShortString()) { 2486 @Override 2487 public void run() { 2488 while (isRunning()) { 2489 processAssignQueue(); 2490 } 2491 pendingAssignQueue.clear(); 2492 } 2493 }; 2494 assignThread.setDaemon(true); 2495 assignThread.start(); 2496 } 2497 2498 private void stopAssignmentThread() { 2499 assignQueueSignal(); 2500 try { 2501 while (assignThread.isAlive()) { 2502 assignQueueSignal(); 2503 assignThread.join(250); 2504 } 2505 } catch (InterruptedException e) { 2506 LOG.warn("join interrupted", e); 2507 Thread.currentThread().interrupt(); 2508 } 2509 } 2510 2511 private void assignQueueSignal() { 2512 assignQueueLock.lock(); 2513 try { 2514 assignQueueFullCond.signal(); 2515 } finally { 2516 assignQueueLock.unlock(); 2517 } 2518 } 2519 2520 @edu.umd.cs.findbugs.annotations.SuppressWarnings("WA_AWAIT_NOT_IN_LOOP") 2521 private HashMap<RegionInfo, RegionStateNode> waitOnAssignQueue() { 2522 HashMap<RegionInfo, RegionStateNode> regions = null; 2523 2524 assignQueueLock.lock(); 2525 try { 2526 if (pendingAssignQueue.isEmpty() && isRunning()) { 2527 assignQueueFullCond.await(); 2528 } 2529 2530 if (!isRunning()) { 2531 return null; 2532 } 2533 assignQueueFullCond.await(assignDispatchWaitMillis, TimeUnit.MILLISECONDS); 2534 regions = new HashMap<RegionInfo, RegionStateNode>(pendingAssignQueue.size()); 2535 for (RegionStateNode regionNode : pendingAssignQueue) { 2536 regions.put(regionNode.getRegionInfo(), regionNode); 2537 } 2538 pendingAssignQueue.clear(); 2539 } catch (InterruptedException e) { 2540 LOG.warn("got interrupted ", e); 2541 Thread.currentThread().interrupt(); 2542 } finally { 2543 assignQueueLock.unlock(); 2544 } 2545 return regions; 2546 } 2547 2548 private void processAssignQueue() { 2549 final HashMap<RegionInfo, RegionStateNode> regions = waitOnAssignQueue(); 2550 if (regions == null || regions.size() == 0 || !isRunning()) { 2551 return; 2552 } 2553 2554 if (LOG.isTraceEnabled()) { 2555 LOG.trace("PROCESS ASSIGN QUEUE regionCount=" + regions.size()); 2556 } 2557 2558 // TODO: Optimize balancer. pass a RegionPlan? 2559 final HashMap<RegionInfo, ServerName> retainMap = new HashMap<>(); 2560 final List<RegionInfo> userHRIs = new ArrayList<>(regions.size()); 2561 // Regions for system tables requiring reassignment 2562 final List<RegionInfo> systemHRIs = new ArrayList<>(); 2563 for (RegionStateNode regionStateNode : regions.values()) { 2564 boolean sysTable = regionStateNode.isSystemTable(); 2565 final List<RegionInfo> hris = sysTable ? systemHRIs : userHRIs; 2566 if (regionStateNode.getRegionLocation() != null) { 2567 retainMap.put(regionStateNode.getRegionInfo(), regionStateNode.getRegionLocation()); 2568 } else { 2569 hris.add(regionStateNode.getRegionInfo()); 2570 } 2571 } 2572 2573 // TODO: connect with the listener to invalidate the cache 2574 2575 // TODO use events 2576 List<ServerName> servers = master.getServerManager().createDestinationServersList(); 2577 for (int i = 0; servers.size() < 1; ++i) { 2578 // Report every fourth time around this loop; try not to flood log. 2579 if (i % 4 == 0) { 2580 LOG.warn("No servers available; cannot place " + regions.size() + " unassigned regions."); 2581 } 2582 2583 if (!isRunning()) { 2584 LOG.debug("Stopped! Dropping assign of " + regions.size() + " queued regions."); 2585 return; 2586 } 2587 Threads.sleep(250); 2588 servers = master.getServerManager().createDestinationServersList(); 2589 } 2590 2591 if (!systemHRIs.isEmpty()) { 2592 // System table regions requiring reassignment are present, get region servers 2593 // not available for system table regions 2594 final List<ServerName> excludeServers = getExcludedServersForSystemTable(); 2595 List<ServerName> serversForSysTables = 2596 servers.stream().filter(s -> !excludeServers.contains(s)).collect(Collectors.toList()); 2597 if (serversForSysTables.isEmpty()) { 2598 LOG.warn("Filtering old server versions and the excluded produced an empty set; " 2599 + "instead considering all candidate servers!"); 2600 } 2601 LOG.debug("Processing assignQueue; systemServersCount=" + serversForSysTables.size() 2602 + ", allServersCount=" + servers.size()); 2603 processAssignmentPlans(regions, null, systemHRIs, 2604 serversForSysTables.isEmpty() && !containsBogusAssignments(regions, systemHRIs) 2605 ? servers 2606 : serversForSysTables); 2607 } 2608 2609 processAssignmentPlans(regions, retainMap, userHRIs, servers); 2610 } 2611 2612 private boolean containsBogusAssignments(Map<RegionInfo, RegionStateNode> regions, 2613 List<RegionInfo> hirs) { 2614 for (RegionInfo ri : hirs) { 2615 if ( 2616 regions.get(ri).getRegionLocation() != null 2617 && regions.get(ri).getRegionLocation().equals(LoadBalancer.BOGUS_SERVER_NAME) 2618 ) { 2619 return true; 2620 } 2621 } 2622 return false; 2623 } 2624 2625 private void processAssignmentPlans(final HashMap<RegionInfo, RegionStateNode> regions, 2626 final HashMap<RegionInfo, ServerName> retainMap, final List<RegionInfo> hris, 2627 final List<ServerName> servers) { 2628 boolean isTraceEnabled = LOG.isTraceEnabled(); 2629 if (isTraceEnabled) { 2630 LOG.trace("Available servers count=" + servers.size() + ": " + servers); 2631 } 2632 2633 final LoadBalancer balancer = getBalancer(); 2634 // ask the balancer where to place regions 2635 if (retainMap != null && !retainMap.isEmpty()) { 2636 if (isTraceEnabled) { 2637 LOG.trace("retain assign regions=" + retainMap); 2638 } 2639 try { 2640 acceptPlan(regions, balancer.retainAssignment(retainMap, servers)); 2641 } catch (IOException e) { 2642 LOG.warn("unable to retain assignment", e); 2643 addToPendingAssignment(regions, retainMap.keySet()); 2644 } 2645 } 2646 2647 // TODO: Do we need to split retain and round-robin? 2648 // the retain seems to fallback to round-robin/random if the region is not in the map. 2649 if (!hris.isEmpty()) { 2650 Collections.sort(hris, RegionInfo.COMPARATOR); 2651 if (isTraceEnabled) { 2652 LOG.trace("round robin regions=" + hris); 2653 } 2654 try { 2655 acceptPlan(regions, balancer.roundRobinAssignment(hris, servers)); 2656 } catch (IOException e) { 2657 LOG.warn("unable to round-robin assignment", e); 2658 addToPendingAssignment(regions, hris); 2659 } 2660 } 2661 } 2662 2663 private void acceptPlan(final HashMap<RegionInfo, RegionStateNode> regions, 2664 final Map<ServerName, List<RegionInfo>> plan) throws HBaseIOException { 2665 final ProcedureEvent<?>[] events = new ProcedureEvent[regions.size()]; 2666 final long st = EnvironmentEdgeManager.currentTime(); 2667 2668 if (plan.isEmpty()) { 2669 throw new HBaseIOException("unable to compute plans for regions=" + regions.size()); 2670 } 2671 2672 int evcount = 0; 2673 for (Map.Entry<ServerName, List<RegionInfo>> entry : plan.entrySet()) { 2674 final ServerName server = entry.getKey(); 2675 for (RegionInfo hri : entry.getValue()) { 2676 final RegionStateNode regionNode = regions.get(hri); 2677 regionNode.setRegionLocation(server); 2678 if (server.equals(LoadBalancer.BOGUS_SERVER_NAME) && regionNode.isSystemTable()) { 2679 assignQueueLock.lock(); 2680 try { 2681 pendingAssignQueue.add(regionNode); 2682 } finally { 2683 assignQueueLock.unlock(); 2684 } 2685 } else { 2686 events[evcount++] = regionNode.getProcedureEvent(); 2687 } 2688 } 2689 } 2690 ProcedureEvent.wakeEvents(getProcedureScheduler(), events); 2691 2692 final long et = EnvironmentEdgeManager.currentTime(); 2693 if (LOG.isTraceEnabled()) { 2694 LOG.trace("ASSIGN ACCEPT " + events.length + " -> " + StringUtils.humanTimeDiff(et - st)); 2695 } 2696 } 2697 2698 private void addToPendingAssignment(final HashMap<RegionInfo, RegionStateNode> regions, 2699 final Collection<RegionInfo> pendingRegions) { 2700 assignQueueLock.lock(); 2701 try { 2702 for (RegionInfo hri : pendingRegions) { 2703 pendingAssignQueue.add(regions.get(hri)); 2704 } 2705 } finally { 2706 assignQueueLock.unlock(); 2707 } 2708 } 2709 2710 /** 2711 * For a given cluster with mixed versions of servers, get a list of servers with lower versions, 2712 * where system table regions should not be assigned to. For system table, we must assign regions 2713 * to a server with highest version. However, we can disable this exclusion using config: 2714 * "hbase.min.version.move.system.tables" if checkForMinVersion is true. Detailed explanation 2715 * available with definition of minVersionToMoveSysTables. 2716 * @return List of Excluded servers for System table regions. 2717 */ 2718 public List<ServerName> getExcludedServersForSystemTable() { 2719 // TODO: This should be a cached list kept by the ServerManager rather than calculated on each 2720 // move or system region assign. The RegionServerTracker keeps list of online Servers with 2721 // RegionServerInfo that includes Version. 2722 List<Pair<ServerName, String>> serverList = 2723 master.getServerManager().getOnlineServersList().stream() 2724 .map(s -> new Pair<>(s, master.getRegionServerVersion(s))).collect(Collectors.toList()); 2725 if (serverList.isEmpty()) { 2726 return new ArrayList<>(); 2727 } 2728 String highestVersion = Collections 2729 .max(serverList, (o1, o2) -> VersionInfo.compareVersion(o1.getSecond(), o2.getSecond())) 2730 .getSecond(); 2731 if (!DEFAULT_MIN_VERSION_MOVE_SYS_TABLES_CONFIG.equals(minVersionToMoveSysTables)) { 2732 int comparedValue = VersionInfo.compareVersion(minVersionToMoveSysTables, highestVersion); 2733 if (comparedValue > 0) { 2734 return new ArrayList<>(); 2735 } 2736 } 2737 return serverList.stream().filter(pair -> !pair.getSecond().equals(highestVersion)) 2738 .map(Pair::getFirst).collect(Collectors.toList()); 2739 } 2740 2741 MasterServices getMaster() { 2742 return master; 2743 } 2744 2745 /** Returns a snapshot of rsReports */ 2746 public Map<ServerName, Set<byte[]>> getRSReports() { 2747 Map<ServerName, Set<byte[]>> rsReportsSnapshot = new HashMap<>(); 2748 synchronized (rsReports) { 2749 rsReports.entrySet().forEach(e -> rsReportsSnapshot.put(e.getKey(), e.getValue())); 2750 } 2751 return rsReportsSnapshot; 2752 } 2753 2754 /** 2755 * Provide regions state count for given table. e.g howmany regions of give table are 2756 * opened/closed/rit etc 2757 * @param tableName TableName 2758 * @return region states count 2759 */ 2760 public RegionStatesCount getRegionStatesCount(TableName tableName) { 2761 int openRegionsCount = 0; 2762 int closedRegionCount = 0; 2763 int ritCount = 0; 2764 int splitRegionCount = 0; 2765 int totalRegionCount = 0; 2766 if (!isTableDisabled(tableName)) { 2767 final List<RegionState> states = regionStates.getTableRegionStates(tableName); 2768 for (RegionState regionState : states) { 2769 if (regionState.isOpened()) { 2770 openRegionsCount++; 2771 } else if (regionState.isClosed()) { 2772 closedRegionCount++; 2773 } else if (regionState.isSplit()) { 2774 splitRegionCount++; 2775 } 2776 } 2777 totalRegionCount = states.size(); 2778 ritCount = totalRegionCount - openRegionsCount - splitRegionCount; 2779 } 2780 return new RegionStatesCount.RegionStatesCountBuilder().setOpenRegions(openRegionsCount) 2781 .setClosedRegions(closedRegionCount).setSplitRegions(splitRegionCount) 2782 .setRegionsInTransition(ritCount).setTotalRegions(totalRegionCount).build(); 2783 } 2784 2785}