View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  
19  package org.apache.hadoop.hbase.regionserver;
20  
21  import org.apache.commons.logging.Log;
22  import org.apache.commons.logging.LogFactory;
23  import org.apache.hadoop.conf.Configuration;
24  import org.apache.hadoop.fs.FileSystem;
25  import org.apache.hadoop.fs.Path;
26  import org.apache.hadoop.hbase.CellScanner;
27  import org.apache.hadoop.hbase.HBaseTestingUtility;
28  import org.apache.hadoop.hbase.HConstants;
29  import org.apache.hadoop.hbase.HRegionInfo;
30  import org.apache.hadoop.hbase.HTableDescriptor;
31  import org.apache.hadoop.hbase.MiniHBaseCluster;
32  import org.apache.hadoop.hbase.TableName;
33  import org.apache.hadoop.hbase.client.Admin;
34  import org.apache.hadoop.hbase.client.Durability;
35  import org.apache.hadoop.hbase.client.Mutation;
36  import org.apache.hadoop.hbase.client.Put;
37  import org.apache.hadoop.hbase.client.Table;
38  import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
39  import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
40  import org.apache.hadoop.hbase.coprocessor.ObserverContext;
41  import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
42  import org.apache.hadoop.hbase.coprocessor.RegionObserver;
43  import org.apache.hadoop.hbase.coprocessor.RegionServerCoprocessorEnvironment;
44  import org.apache.hadoop.hbase.coprocessor.RegionServerObserver;
45  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos;
46  import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
47  import org.apache.hadoop.hbase.replication.ReplicationEndpoint;
48  import org.apache.hadoop.hbase.testclassification.MediumTests;
49  import org.apache.hadoop.hbase.testclassification.RegionServerTests;
50  import org.apache.hadoop.hbase.util.Bytes;
51  import org.apache.hadoop.hbase.util.JVMClusterUtil;
52  import org.apache.hadoop.hbase.wal.WAL;
53  import org.apache.hadoop.hdfs.DFSConfigKeys;
54  import org.apache.hadoop.hdfs.MiniDFSCluster;
55  import org.junit.After;
56  import org.junit.Before;
57  import org.junit.Test;
58  import org.junit.experimental.categories.Category;
59  
60  import java.io.IOException;
61  import java.util.Collection;
62  import java.util.List;
63  
64  import static org.junit.Assert.assertFalse;
65  import static org.junit.Assert.assertNotNull;
66  import static org.junit.Assert.assertTrue;
67  
68  /**
69   * Tests around regionserver shutdown and abort
70   */
71  @Category({RegionServerTests.class, MediumTests.class})
72  public class TestRegionServerAbort {
73    private static final byte[] FAMILY_BYTES = Bytes.toBytes("f");
74  
75    private static final Log LOG = LogFactory.getLog(TestRegionServerAbort.class);
76  
77    private HBaseTestingUtility testUtil;
78    private Configuration conf;
79    private MiniDFSCluster dfsCluster;
80    private MiniHBaseCluster cluster;
81  
82    @Before
83    public void setup() throws Exception {
84      testUtil = new HBaseTestingUtility();
85      conf = testUtil.getConfiguration();
86      conf.set(CoprocessorHost.REGIONSERVER_COPROCESSOR_CONF_KEY,
87          StopBlockingRegionObserver.class.getName());
88      conf.set(CoprocessorHost.REGION_COPROCESSOR_CONF_KEY,
89          StopBlockingRegionObserver.class.getName());
90      // make sure we have multiple blocks so that the client does not prefetch all block locations
91      conf.set("dfs.blocksize", Long.toString(100 * 1024));
92      // prefetch the first block
93      conf.set(DFSConfigKeys.DFS_CLIENT_READ_PREFETCH_SIZE_KEY, Long.toString(100 * 1024));
94  
95      testUtil.startMiniZKCluster();
96      dfsCluster = testUtil.startMiniDFSCluster(2);
97      cluster = testUtil.startMiniHBaseCluster(1, 2);
98    }
99  
100   @After
101   public void tearDown() throws Exception {
102     for (JVMClusterUtil.RegionServerThread t : cluster.getRegionServerThreads()) {
103       HRegionServer rs = t.getRegionServer();
104       RegionServerCoprocessorHost cpHost = rs.getRegionServerCoprocessorHost();
105       StopBlockingRegionObserver cp = (StopBlockingRegionObserver)
106           cpHost.findCoprocessor(StopBlockingRegionObserver.class.getName());
107       cp.setStopAllowed(true);
108     }
109     testUtil.shutdownMiniCluster();
110   }
111 
112   /**
113    * Test that a regionserver is able to abort properly, even when a coprocessor
114    * throws an exception in preStopRegionServer().
115    */
116   @Test
117   public void testAbortFromRPC() throws Exception {
118     TableName tableName = TableName.valueOf("testAbortFromRPC");
119     // create a test table
120     Table table = testUtil.createTable(tableName, FAMILY_BYTES);
121 
122     // write some edits
123     testUtil.loadTable(table, FAMILY_BYTES);
124     LOG.info("Wrote data");
125     // force a flush
126     cluster.flushcache(tableName);
127     LOG.info("Flushed table");
128 
129     // Send a poisoned put to trigger the abort
130     Put put = new Put(new byte[]{0, 0, 0, 0});
131     put.addColumn(FAMILY_BYTES, Bytes.toBytes("c"), new byte[]{});
132     put.setAttribute(StopBlockingRegionObserver.DO_ABORT, new byte[]{1});
133 
134     table.put(put);
135     // should have triggered an abort due to FileNotFoundException
136 
137     // verify that the regionserver is stopped
138     HRegion firstRegion = cluster.findRegionsForTable(tableName).get(0);
139     assertNotNull(firstRegion);
140     assertNotNull(firstRegion.getRegionServerServices());
141     LOG.info("isAborted = " + firstRegion.getRegionServerServices().isAborted());
142     assertTrue(firstRegion.getRegionServerServices().isAborted());
143     LOG.info("isStopped = " + firstRegion.getRegionServerServices().isStopped());
144     assertTrue(firstRegion.getRegionServerServices().isStopped());
145   }
146 
147   /**
148    * Test that a coprocessor is able to override a normal regionserver stop request.
149    */
150   @Test
151   public void testStopOverrideFromCoprocessor() throws Exception {
152     Admin admin = testUtil.getHBaseAdmin();
153     HRegionServer regionserver = cluster.getRegionServer(0);
154     admin.stopRegionServer(regionserver.getServerName().getHostAndPort());
155 
156     // regionserver should have failed to stop due to coprocessor
157     assertFalse(cluster.getRegionServer(0).isAborted());
158     assertFalse(cluster.getRegionServer(0).isStopped());
159   }
160 
161   public static class StopBlockingRegionObserver extends BaseRegionObserver
162       implements RegionServerObserver {
163     public static final String DO_ABORT = "DO_ABORT";
164     private boolean stopAllowed;
165 
166     @Override
167     public void prePut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit,
168                        Durability durability) throws IOException {
169       if (put.getAttribute(DO_ABORT) != null) {
170         HRegionServer rs = (HRegionServer) c.getEnvironment().getRegionServerServices();
171         LOG.info("Triggering abort for regionserver " + rs.getServerName());
172         rs.abort("Aborting for test");
173       }
174     }
175 
176     @Override
177     public void preStopRegionServer(ObserverContext<RegionServerCoprocessorEnvironment> env)
178         throws IOException {
179       if (!stopAllowed) {
180         throw new IOException("Stop not allowed");
181       }
182     }
183 
184     @Override
185     public void preMerge(ObserverContext<RegionServerCoprocessorEnvironment> ctx,
186                          Region regionA, Region regionB) throws IOException {
187       // no-op
188     }
189 
190     @Override
191     public void postMerge(ObserverContext<RegionServerCoprocessorEnvironment> c,
192                           Region regionA, Region regionB, Region mergedRegion) throws IOException {
193       // no-op
194     }
195 
196     @Override
197     public void preMergeCommit(ObserverContext<RegionServerCoprocessorEnvironment> ctx,
198                                Region regionA, Region regionB, List<Mutation> metaEntries)
199         throws IOException {
200       // no-op
201     }
202 
203     @Override
204     public void postMergeCommit(ObserverContext<RegionServerCoprocessorEnvironment> ctx,
205                                 Region regionA, Region regionB, Region mergedRegion)
206         throws IOException {
207       // no-op
208     }
209 
210     @Override
211     public void preRollBackMerge(ObserverContext<RegionServerCoprocessorEnvironment> ctx,
212                                  Region regionA, Region regionB) throws IOException {
213       // no-op
214     }
215 
216     @Override
217     public void postRollBackMerge(ObserverContext<RegionServerCoprocessorEnvironment> ctx,
218                                   Region regionA, Region regionB) throws IOException {
219       // no-op
220     }
221 
222     @Override
223     public void preRollWALWriterRequest(ObserverContext<RegionServerCoprocessorEnvironment> ctx)
224         throws IOException {
225       // no-op
226     }
227 
228     @Override
229     public void postRollWALWriterRequest(ObserverContext<RegionServerCoprocessorEnvironment> ctx)
230         throws IOException {
231       // no-op
232     }
233 
234     @Override
235     public ReplicationEndpoint postCreateReplicationEndPoint(
236         ObserverContext<RegionServerCoprocessorEnvironment> ctx, ReplicationEndpoint endpoint) {
237       return null;
238     }
239 
240     @Override
241     public void preReplicateLogEntries(ObserverContext<RegionServerCoprocessorEnvironment> ctx,
242                                        List<AdminProtos.WALEntry> entries, CellScanner cells)
243         throws IOException {
244       // no-op
245     }
246 
247     @Override
248     public void postReplicateLogEntries(ObserverContext<RegionServerCoprocessorEnvironment> ctx,
249                                         List<AdminProtos.WALEntry> entries, CellScanner cells)
250         throws IOException {
251       // no-op
252     }
253 
254     public void setStopAllowed(boolean allowed) {
255       this.stopAllowed = allowed;
256     }
257 
258     public boolean isStopAllowed() {
259       return stopAllowed;
260     }
261   }
262 }