1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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
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
91 conf.set("dfs.blocksize", Long.toString(100 * 1024));
92
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
114
115
116 @Test
117 public void testAbortFromRPC() throws Exception {
118 TableName tableName = TableName.valueOf("testAbortFromRPC");
119
120 Table table = testUtil.createTable(tableName, FAMILY_BYTES);
121
122
123 testUtil.loadTable(table, FAMILY_BYTES);
124 LOG.info("Wrote data");
125
126 cluster.flushcache(tableName);
127 LOG.info("Flushed table");
128
129
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
136
137
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
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
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
188 }
189
190 @Override
191 public void postMerge(ObserverContext<RegionServerCoprocessorEnvironment> c,
192 Region regionA, Region regionB, Region mergedRegion) throws IOException {
193
194 }
195
196 @Override
197 public void preMergeCommit(ObserverContext<RegionServerCoprocessorEnvironment> ctx,
198 Region regionA, Region regionB, List<Mutation> metaEntries)
199 throws IOException {
200
201 }
202
203 @Override
204 public void postMergeCommit(ObserverContext<RegionServerCoprocessorEnvironment> ctx,
205 Region regionA, Region regionB, Region mergedRegion)
206 throws IOException {
207
208 }
209
210 @Override
211 public void preRollBackMerge(ObserverContext<RegionServerCoprocessorEnvironment> ctx,
212 Region regionA, Region regionB) throws IOException {
213
214 }
215
216 @Override
217 public void postRollBackMerge(ObserverContext<RegionServerCoprocessorEnvironment> ctx,
218 Region regionA, Region regionB) throws IOException {
219
220 }
221
222 @Override
223 public void preRollWALWriterRequest(ObserverContext<RegionServerCoprocessorEnvironment> ctx)
224 throws IOException {
225
226 }
227
228 @Override
229 public void postRollWALWriterRequest(ObserverContext<RegionServerCoprocessorEnvironment> ctx)
230 throws IOException {
231
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
245 }
246
247 @Override
248 public void postReplicateLogEntries(ObserverContext<RegionServerCoprocessorEnvironment> ctx,
249 List<AdminProtos.WALEntry> entries, CellScanner cells)
250 throws IOException {
251
252 }
253
254 public void setStopAllowed(boolean allowed) {
255 this.stopAllowed = allowed;
256 }
257
258 public boolean isStopAllowed() {
259 return stopAllowed;
260 }
261 }
262 }