001/** 002 * 003 * Licensed to the Apache Software Foundation (ASF) under one 004 * or more contributor license agreements. See the NOTICE file 005 * distributed with this work for additional information 006 * regarding copyright ownership. The ASF licenses this file 007 * to you under the Apache License, Version 2.0 (the 008 * "License"); you may not use this file except in compliance 009 * with the License. You may obtain a copy of the License at 010 * 011 * http://www.apache.org/licenses/LICENSE-2.0 012 * 013 * Unless required by applicable law or agreed to in writing, software 014 * distributed under the License is distributed on an "AS IS" BASIS, 015 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 016 * See the License for the specific language governing permissions and 017 * limitations under the License. 018 */ 019package org.apache.hadoop.hbase.util; 020 021import java.io.InterruptedIOException; 022import java.io.IOException; 023import java.lang.reflect.Constructor; 024import java.lang.reflect.InvocationTargetException; 025import java.util.List; 026 027import org.apache.yetus.audience.InterfaceAudience; 028import org.slf4j.Logger; 029import org.slf4j.LoggerFactory; 030import org.apache.hadoop.conf.Configuration; 031import org.apache.hadoop.hbase.CoordinatedStateManager; 032import org.apache.hadoop.hbase.master.HMaster; 033import org.apache.hadoop.hbase.regionserver.HRegionServer; 034 035/** 036 * Utility used running a cluster all in the one JVM. 037 */ 038@InterfaceAudience.Private 039public class JVMClusterUtil { 040 private static final Logger LOG = LoggerFactory.getLogger(JVMClusterUtil.class); 041 042 /** 043 * Datastructure to hold RegionServer Thread and RegionServer instance 044 */ 045 public static class RegionServerThread extends Thread { 046 private final HRegionServer regionServer; 047 048 public RegionServerThread(final HRegionServer r, final int index) { 049 super(r, "RS:" + index + ";" + r.getServerName().toShortString()); 050 this.regionServer = r; 051 } 052 053 /** @return the region server */ 054 public HRegionServer getRegionServer() { 055 return this.regionServer; 056 } 057 058 /** 059 * Block until the region server has come online, indicating it is ready 060 * to be used. 061 */ 062 public void waitForServerOnline() { 063 // The server is marked online after the init method completes inside of 064 // the HRS#run method. HRS#init can fail for whatever region. In those 065 // cases, we'll jump out of the run without setting online flag. Check 066 // stopRequested so we don't wait here a flag that will never be flipped. 067 regionServer.waitForServerOnline(); 068 } 069 } 070 071 /** 072 * Creates a {@link RegionServerThread}. 073 * Call 'start' on the returned thread to make it run. 074 * @param c Configuration to use. 075 * @param hrsc Class to create. 076 * @param index Used distinguishing the object returned. 077 * @throws IOException 078 * @return Region server added. 079 */ 080 public static JVMClusterUtil.RegionServerThread createRegionServerThread(final Configuration c, 081 final Class<? extends HRegionServer> hrsc, final int index) throws IOException { 082 HRegionServer server; 083 try { 084 Constructor<? extends HRegionServer> ctor = hrsc.getConstructor(Configuration.class); 085 ctor.setAccessible(true); 086 server = ctor.newInstance(c); 087 } catch (InvocationTargetException ite) { 088 Throwable target = ite.getTargetException(); 089 throw new RuntimeException("Failed construction of RegionServer: " + 090 hrsc.toString() + ((target.getCause() != null)? 091 target.getCause().getMessage(): ""), target); 092 } catch (Exception e) { 093 IOException ioe = new IOException(); 094 ioe.initCause(e); 095 throw ioe; 096 } 097 return new JVMClusterUtil.RegionServerThread(server, index); 098 } 099 100 101 /** 102 * Datastructure to hold Master Thread and Master instance 103 */ 104 public static class MasterThread extends Thread { 105 private final HMaster master; 106 107 public MasterThread(final HMaster m, final int index) { 108 super(m, "M:" + index + ";" + m.getServerName().toShortString()); 109 this.master = m; 110 } 111 112 /** @return the master */ 113 public HMaster getMaster() { 114 return this.master; 115 } 116 } 117 118 /** 119 * Creates a {@link MasterThread}. 120 * Call 'start' on the returned thread to make it run. 121 * @param c Configuration to use. 122 * @param hmc Class to create. 123 * @param index Used distinguishing the object returned. 124 * @throws IOException 125 * @return Master added. 126 */ 127 public static JVMClusterUtil.MasterThread createMasterThread(final Configuration c, 128 final Class<? extends HMaster> hmc, final int index) throws IOException { 129 HMaster server; 130 try { 131 server = hmc.getConstructor(Configuration.class).newInstance(c); 132 } catch (InvocationTargetException ite) { 133 Throwable target = ite.getTargetException(); 134 throw new RuntimeException("Failed construction of Master: " + 135 hmc.toString() + ((target.getCause() != null)? 136 target.getCause().getMessage(): ""), target); 137 } catch (Exception e) { 138 IOException ioe = new IOException(); 139 ioe.initCause(e); 140 throw ioe; 141 } 142 return new JVMClusterUtil.MasterThread(server, index); 143 } 144 145 private static JVMClusterUtil.MasterThread findActiveMaster( 146 List<JVMClusterUtil.MasterThread> masters) { 147 for (JVMClusterUtil.MasterThread t : masters) { 148 if (t.master.isActiveMaster()) { 149 return t; 150 } 151 } 152 153 return null; 154 } 155 156 /** 157 * Start the cluster. Waits until there is a primary master initialized 158 * and returns its address. 159 * @param masters 160 * @param regionservers 161 * @return Address to use contacting primary master. 162 */ 163 public static String startup(final List<JVMClusterUtil.MasterThread> masters, 164 final List<JVMClusterUtil.RegionServerThread> regionservers) throws IOException { 165 166 Configuration configuration = null; 167 168 if (masters == null || masters.isEmpty()) { 169 return null; 170 } 171 172 for (JVMClusterUtil.MasterThread t : masters) { 173 configuration = t.getMaster().getConfiguration(); 174 t.start(); 175 } 176 177 // Wait for an active master 178 // having an active master before starting the region threads allows 179 // then to succeed on their connection to master 180 long startTime = System.currentTimeMillis(); 181 while (findActiveMaster(masters) == null) { 182 try { 183 Thread.sleep(100); 184 } catch (InterruptedException e) { 185 throw (InterruptedIOException)new InterruptedIOException().initCause(e); 186 } 187 int startTimeout = configuration != null ? Integer.parseInt( 188 configuration.get("hbase.master.start.timeout.localHBaseCluster", "30000")) : 30000; 189 if (System.currentTimeMillis() > startTime + startTimeout) { 190 String msg = "Master not active after " + startTimeout + "ms"; 191 Threads.printThreadInfo(System.out, "Thread dump because: " + msg); 192 throw new RuntimeException(msg); 193 } 194 } 195 196 if (regionservers != null) { 197 for (JVMClusterUtil.RegionServerThread t: regionservers) { 198 t.start(); 199 } 200 } 201 202 // Wait for an active master to be initialized (implies being master) 203 // with this, when we return the cluster is complete 204 startTime = System.currentTimeMillis(); 205 final int maxwait = 200000; 206 while (true) { 207 JVMClusterUtil.MasterThread t = findActiveMaster(masters); 208 if (t != null && t.master.isInitialized()) { 209 return t.master.getServerName().toString(); 210 } 211 // REMOVE 212 if (System.currentTimeMillis() > startTime + 10000) { 213 try { 214 Thread.sleep(1000); 215 } catch (InterruptedException e) { 216 throw (InterruptedIOException)new InterruptedIOException().initCause(e); 217 } 218 } 219 if (System.currentTimeMillis() > startTime + maxwait) { 220 String msg = "Master not initialized after " + maxwait + "ms seconds"; 221 Threads.printThreadInfo(System.out, "Thread dump because: " + msg); 222 throw new RuntimeException(msg); 223 } 224 try { 225 Thread.sleep(100); 226 } catch (InterruptedException e) { 227 throw (InterruptedIOException)new InterruptedIOException().initCause(e); 228 } 229 } 230 } 231 232 /** 233 * @param masters 234 * @param regionservers 235 */ 236 public static void shutdown(final List<MasterThread> masters, 237 final List<RegionServerThread> regionservers) { 238 LOG.debug("Shutting down HBase Cluster"); 239 if (masters != null) { 240 // Do backups first. 241 JVMClusterUtil.MasterThread activeMaster = null; 242 for (JVMClusterUtil.MasterThread t : masters) { 243 if (!t.master.isActiveMaster()) { 244 try { 245 t.master.stopMaster(); 246 } catch (IOException e) { 247 LOG.error("Exception occurred while stopping master", e); 248 } 249 } else { 250 activeMaster = t; 251 } 252 } 253 // Do active after. 254 if (activeMaster != null) { 255 try { 256 activeMaster.master.shutdown(); 257 } catch (IOException e) { 258 LOG.error("Exception occurred in HMaster.shutdown()", e); 259 } 260 } 261 } 262 boolean wasInterrupted = false; 263 final long maxTime = System.currentTimeMillis() + 30 * 1000; 264 if (regionservers != null) { 265 // first try nicely. 266 for (RegionServerThread t : regionservers) { 267 t.getRegionServer().stop("Shutdown requested"); 268 } 269 for (RegionServerThread t : regionservers) { 270 long now = System.currentTimeMillis(); 271 if (t.isAlive() && !wasInterrupted && now < maxTime) { 272 try { 273 t.join(maxTime - now); 274 } catch (InterruptedException e) { 275 LOG.info("Got InterruptedException on shutdown - " + 276 "not waiting anymore on region server ends", e); 277 wasInterrupted = true; // someone wants us to speed up. 278 } 279 } 280 } 281 282 // Let's try to interrupt the remaining threads if any. 283 for (int i = 0; i < 100; ++i) { 284 boolean atLeastOneLiveServer = false; 285 for (RegionServerThread t : regionservers) { 286 if (t.isAlive()) { 287 atLeastOneLiveServer = true; 288 try { 289 LOG.warn("RegionServerThreads remaining, give one more chance before interrupting"); 290 t.join(1000); 291 } catch (InterruptedException e) { 292 wasInterrupted = true; 293 } 294 } 295 } 296 if (!atLeastOneLiveServer) break; 297 for (RegionServerThread t : regionservers) { 298 if (t.isAlive()) { 299 LOG.warn("RegionServerThreads taking too long to stop, interrupting; thread dump " + 300 "if > 3 attempts: i=" + i); 301 if (i > 3) { 302 Threads.printThreadInfo(System.out, "Thread dump " + t.getName()); 303 } 304 t.interrupt(); 305 } 306 } 307 } 308 } 309 310 if (masters != null) { 311 for (JVMClusterUtil.MasterThread t : masters) { 312 while (t.master.isAlive() && !wasInterrupted) { 313 try { 314 // The below has been replaced to debug sometime hangs on end of 315 // tests. 316 // this.master.join(): 317 Threads.threadDumpingIsAlive(t.master.getThread()); 318 } catch(InterruptedException e) { 319 LOG.info("Got InterruptedException on shutdown - " + 320 "not waiting anymore on master ends", e); 321 wasInterrupted = true; 322 } 323 } 324 } 325 } 326 LOG.info("Shutdown of " + 327 ((masters != null) ? masters.size() : "0") + " master(s) and " + 328 ((regionservers != null) ? regionservers.size() : "0") + 329 " regionserver(s) " + (wasInterrupted ? "interrupted" : "complete")); 330 331 if (wasInterrupted){ 332 Thread.currentThread().interrupt(); 333 } 334 } 335}