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.regionserver; 019 020import static org.junit.Assert.assertEquals; 021import static org.junit.Assert.assertFalse; 022import static org.junit.Assert.assertNotNull; 023import static org.junit.Assert.assertTrue; 024 025import java.io.IOException; 026import java.util.List; 027import java.util.Optional; 028import java.util.concurrent.CompletableFuture; 029import java.util.concurrent.ExecutorService; 030import java.util.concurrent.Executors; 031import java.util.concurrent.atomic.AtomicInteger; 032import org.apache.hadoop.conf.Configuration; 033import org.apache.hadoop.fs.FileSystem; 034import org.apache.hadoop.fs.Path; 035import org.apache.hadoop.hbase.HBaseClassTestRule; 036import org.apache.hadoop.hbase.HBaseTestingUtil; 037import org.apache.hadoop.hbase.HConstants; 038import org.apache.hadoop.hbase.SingleProcessHBaseCluster; 039import org.apache.hadoop.hbase.StartTestingClusterOption; 040import org.apache.hadoop.hbase.TableName; 041import org.apache.hadoop.hbase.Waiter; 042import org.apache.hadoop.hbase.client.Admin; 043import org.apache.hadoop.hbase.client.Durability; 044import org.apache.hadoop.hbase.client.Put; 045import org.apache.hadoop.hbase.client.RegionInfo; 046import org.apache.hadoop.hbase.client.Table; 047import org.apache.hadoop.hbase.client.TableDescriptor; 048import org.apache.hadoop.hbase.coprocessor.CoprocessorHost; 049import org.apache.hadoop.hbase.coprocessor.CoreCoprocessor; 050import org.apache.hadoop.hbase.coprocessor.HasRegionServerServices; 051import org.apache.hadoop.hbase.coprocessor.ObserverContext; 052import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor; 053import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; 054import org.apache.hadoop.hbase.coprocessor.RegionObserver; 055import org.apache.hadoop.hbase.coprocessor.RegionServerCoprocessor; 056import org.apache.hadoop.hbase.coprocessor.RegionServerCoprocessorEnvironment; 057import org.apache.hadoop.hbase.coprocessor.RegionServerObserver; 058import org.apache.hadoop.hbase.testclassification.MediumTests; 059import org.apache.hadoop.hbase.testclassification.RegionServerTests; 060import org.apache.hadoop.hbase.util.Bytes; 061import org.apache.hadoop.hbase.util.JVMClusterUtil; 062import org.apache.hadoop.hbase.wal.WAL; 063import org.apache.hadoop.hbase.wal.WALEdit; 064import org.apache.hadoop.hdfs.DFSConfigKeys; 065import org.apache.hadoop.hdfs.MiniDFSCluster; 066import org.junit.After; 067import org.junit.Before; 068import org.junit.ClassRule; 069import org.junit.Test; 070import org.junit.experimental.categories.Category; 071import org.slf4j.Logger; 072import org.slf4j.LoggerFactory; 073 074/** 075 * Tests around regionserver shutdown and abort 076 */ 077@Category({ RegionServerTests.class, MediumTests.class }) 078public class TestRegionServerAbort { 079 080 @ClassRule 081 public static final HBaseClassTestRule CLASS_RULE = 082 HBaseClassTestRule.forClass(TestRegionServerAbort.class); 083 084 private static final byte[] FAMILY_BYTES = Bytes.toBytes("f"); 085 086 private static final Logger LOG = LoggerFactory.getLogger(TestRegionServerAbort.class); 087 088 private HBaseTestingUtil testUtil; 089 private Configuration conf; 090 private MiniDFSCluster dfsCluster; 091 private SingleProcessHBaseCluster cluster; 092 093 @Before 094 public void setup() throws Exception { 095 testUtil = new HBaseTestingUtil(); 096 conf = testUtil.getConfiguration(); 097 conf.set(CoprocessorHost.REGIONSERVER_COPROCESSOR_CONF_KEY, 098 StopBlockingRegionObserver.class.getName()); 099 conf.set(CoprocessorHost.REGION_COPROCESSOR_CONF_KEY, 100 StopBlockingRegionObserver.class.getName()); 101 // make sure we have multiple blocks so that the client does not prefetch all block locations 102 conf.set("dfs.blocksize", Long.toString(100 * 1024)); 103 // prefetch the first block 104 conf.set(DFSConfigKeys.DFS_CLIENT_READ_PREFETCH_SIZE_KEY, Long.toString(100 * 1024)); 105 conf.set(HConstants.REGION_IMPL, ErrorThrowingHRegion.class.getName()); 106 107 testUtil.startMiniZKCluster(); 108 dfsCluster = testUtil.startMiniDFSCluster(2); 109 StartTestingClusterOption option = 110 StartTestingClusterOption.builder().numRegionServers(2).build(); 111 cluster = testUtil.startMiniHBaseCluster(option); 112 } 113 114 @After 115 public void tearDown() throws Exception { 116 String className = StopBlockingRegionObserver.class.getName(); 117 for (JVMClusterUtil.RegionServerThread t : cluster.getRegionServerThreads()) { 118 HRegionServer rs = t.getRegionServer(); 119 RegionServerCoprocessorHost cpHost = rs.getRegionServerCoprocessorHost(); 120 StopBlockingRegionObserver cp = 121 (StopBlockingRegionObserver) cpHost.findCoprocessor(className); 122 cp.setStopAllowed(true); 123 } 124 testUtil.shutdownMiniCluster(); 125 } 126 127 /** 128 * Test that a regionserver is able to abort properly, even when a coprocessor throws an exception 129 * in preStopRegionServer(). 130 */ 131 @Test 132 public void testAbortFromRPC() throws Exception { 133 TableName tableName = TableName.valueOf("testAbortFromRPC"); 134 // create a test table 135 Table table = testUtil.createTable(tableName, FAMILY_BYTES); 136 137 // write some edits 138 testUtil.loadTable(table, FAMILY_BYTES); 139 LOG.info("Wrote data"); 140 // force a flush 141 cluster.flushcache(tableName); 142 LOG.info("Flushed table"); 143 144 // Send a poisoned put to trigger the abort 145 Put put = new Put(new byte[] { 0, 0, 0, 0 }); 146 put.addColumn(FAMILY_BYTES, Bytes.toBytes("c"), new byte[] {}); 147 put.setAttribute(StopBlockingRegionObserver.DO_ABORT, new byte[] { 1 }); 148 149 List<HRegion> regions = cluster.findRegionsForTable(tableName); 150 HRegion firstRegion = cluster.findRegionsForTable(tableName).get(0); 151 table.put(put); 152 // Verify that the regionserver is stopped 153 assertNotNull(firstRegion); 154 assertNotNull(firstRegion.getRegionServerServices()); 155 LOG.info("isAborted = " + firstRegion.getRegionServerServices().isAborted()); 156 assertTrue(firstRegion.getRegionServerServices().isAborted()); 157 LOG.info("isStopped = " + firstRegion.getRegionServerServices().isStopped()); 158 assertTrue(firstRegion.getRegionServerServices().isStopped()); 159 } 160 161 /** 162 * Test that a coprocessor is able to override a normal regionserver stop request. 163 */ 164 @Test 165 public void testStopOverrideFromCoprocessor() throws Exception { 166 Admin admin = testUtil.getAdmin(); 167 HRegionServer regionserver = cluster.getRegionServer(0); 168 admin.stopRegionServer(regionserver.getServerName().getAddress().toString()); 169 170 // regionserver should have failed to stop due to coprocessor 171 assertFalse(cluster.getRegionServer(0).isAborted()); 172 assertFalse(cluster.getRegionServer(0).isStopped()); 173 } 174 175 /** 176 * Tests that only a single abort is processed when multiple aborts are requested. 177 */ 178 @Test 179 public void testMultiAbort() { 180 assertTrue(cluster.getRegionServerThreads().size() > 0); 181 JVMClusterUtil.RegionServerThread t = cluster.getRegionServerThreads().get(0); 182 assertTrue(t.isAlive()); 183 HRegionServer rs = t.getRegionServer(); 184 assertFalse(rs.isAborted()); 185 RegionServerCoprocessorHost cpHost = rs.getRegionServerCoprocessorHost(); 186 StopBlockingRegionObserver cp = (StopBlockingRegionObserver) cpHost 187 .findCoprocessor(StopBlockingRegionObserver.class.getName()); 188 // Enable clean abort. 189 cp.setStopAllowed(true); 190 // Issue two aborts in quick succession. 191 // We need a thread pool here, otherwise the abort() runs into SecurityException when running 192 // from the fork join pool when setting the context classloader. 193 ExecutorService executor = Executors.newFixedThreadPool(2); 194 try { 195 CompletableFuture.runAsync(() -> rs.abort("Abort 1"), executor); 196 CompletableFuture.runAsync(() -> rs.abort("Abort 2"), executor); 197 long testTimeoutMs = 10 * 1000; 198 Waiter.waitFor(cluster.getConf(), testTimeoutMs, (Waiter.Predicate<Exception>) rs::isStopped); 199 // Make sure only one abort is received. 200 assertEquals(1, cp.getNumAbortsRequested()); 201 } finally { 202 executor.shutdownNow(); 203 } 204 } 205 206 @CoreCoprocessor 207 public static class StopBlockingRegionObserver 208 implements RegionServerCoprocessor, RegionCoprocessor, RegionServerObserver, RegionObserver { 209 public static final String DO_ABORT = "DO_ABORT"; 210 private boolean stopAllowed; 211 private AtomicInteger abortCount = new AtomicInteger(); 212 213 @Override 214 public Optional<RegionObserver> getRegionObserver() { 215 return Optional.of(this); 216 } 217 218 @Override 219 public Optional<RegionServerObserver> getRegionServerObserver() { 220 return Optional.of(this); 221 } 222 223 @Override 224 public void prePut(ObserverContext<? extends RegionCoprocessorEnvironment> c, Put put, 225 WALEdit edit, Durability durability) throws IOException { 226 if (put.getAttribute(DO_ABORT) != null) { 227 // TODO: Change this so it throws a CP Abort Exception instead. 228 RegionServerServices rss = 229 ((HasRegionServerServices) c.getEnvironment()).getRegionServerServices(); 230 String str = "Aborting for test"; 231 LOG.info(str + " " + rss.getServerName()); 232 rss.abort(str, new Throwable(str)); 233 } 234 } 235 236 @Override 237 public void preStopRegionServer(ObserverContext<RegionServerCoprocessorEnvironment> env) 238 throws IOException { 239 abortCount.incrementAndGet(); 240 if (!stopAllowed) { 241 throw new IOException("Stop not allowed"); 242 } 243 } 244 245 public int getNumAbortsRequested() { 246 return abortCount.get(); 247 } 248 249 public void setStopAllowed(boolean allowed) { 250 this.stopAllowed = allowed; 251 } 252 } 253 254 /** 255 * Throws an exception during store file refresh in order to trigger a regionserver abort. 256 */ 257 public static class ErrorThrowingHRegion extends HRegion { 258 public ErrorThrowingHRegion(Path tableDir, WAL wal, FileSystem fs, Configuration confParam, 259 RegionInfo regionInfo, TableDescriptor htd, RegionServerServices rsServices) { 260 super(tableDir, wal, fs, confParam, regionInfo, htd, rsServices); 261 } 262 263 public ErrorThrowingHRegion(HRegionFileSystem fs, WAL wal, Configuration confParam, 264 TableDescriptor htd, RegionServerServices rsServices) { 265 super(fs, wal, confParam, htd, rsServices); 266 } 267 268 @Override 269 protected boolean refreshStoreFiles(boolean force) throws IOException { 270 // forced when called through RegionScannerImpl.handleFileNotFound() 271 if (force) { 272 throw new IOException("Failing file refresh for testing"); 273 } 274 return super.refreshStoreFiles(force); 275 } 276 } 277}