1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 package org.apache.hadoop.hbase.client;
21
22 import static org.junit.Assert.assertEquals;
23 import static org.junit.Assert.assertTrue;
24
25 import java.io.IOException;
26 import java.util.ArrayList;
27 import java.util.List;
28 import java.util.Set;
29 import java.util.concurrent.atomic.AtomicBoolean;
30
31 import org.apache.hadoop.hbase.Cell;
32 import org.apache.hadoop.hbase.CellUtil;
33 import org.apache.hadoop.hbase.HBaseTestingUtility;
34 import org.apache.hadoop.hbase.HColumnDescriptor;
35 import org.apache.hadoop.hbase.HConstants;
36 import org.apache.hadoop.hbase.HRegionInfo;
37 import org.apache.hadoop.hbase.HTableDescriptor;
38 import org.apache.hadoop.hbase.TableName;
39 import org.apache.hadoop.hbase.regionserver.wal.FailedLogCloseException;
40 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
41 import org.apache.hadoop.hbase.regionserver.wal.WALCoprocessorHost;
42 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
43 import org.apache.hadoop.hbase.testclassification.SmallTests;
44 import org.apache.hadoop.hbase.util.Bytes;
45 import org.apache.hadoop.hbase.wal.DefaultWALProvider;
46 import org.apache.hadoop.hbase.wal.WAL;
47 import org.apache.hadoop.hbase.wal.WALFactory;
48 import org.apache.hadoop.hbase.wal.WALKey;
49 import org.junit.AfterClass;
50 import org.junit.Assert;
51 import org.junit.BeforeClass;
52 import org.junit.Rule;
53 import org.junit.Test;
54 import org.junit.experimental.categories.Category;
55 import org.junit.rules.TestName;
56
57 @Category(SmallTests.class)
58 public class TestRollbackFromClient {
59 @Rule
60 public TestName name = new TestName();
61 private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
62 private static final byte[] FAMILY = Bytes.toBytes("testFamily");
63 private static final int SLAVES = 3;
64 private static final byte[] ROW = Bytes.toBytes("testRow");
65 private static final byte[] QUALIFIER = Bytes.toBytes("testQualifier");
66 private static final byte[] QUALIFIER_V2 = Bytes.toBytes("testQualifierV2");
67 private static final byte[] VALUE = Bytes.toBytes("testValue");
68
69 @BeforeClass
70 public static void setUpBeforeClass() throws Exception {
71 TEST_UTIL.getConfiguration().setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 2);
72 TEST_UTIL.getConfiguration().set(WALFactory.WAL_PROVIDER,
73 FailedDefaultWALProvider.class.getName());
74 TEST_UTIL.startMiniCluster(SLAVES);
75 }
76
77 @AfterClass
78 public static void tearDownAfterClass() throws Exception {
79 TEST_UTIL.shutdownMiniCluster();
80 }
81
82 @Test
83 public void testAppendRollback() throws IOException {
84 Updater updateForEmptyTable = new Updater() {
85 @Override
86 public int updateData(Table table, byte[] family) {
87 try {
88 Append append = new Append(ROW);
89 append.add(FAMILY, QUALIFIER, VALUE);
90 append.add(FAMILY, QUALIFIER_V2, VALUE);
91 FailedHLog.SHOULD_FAIL.set(true);
92 table.append(append);
93 } catch (IOException e) {
94
95 } finally {
96 FailedHLog.SHOULD_FAIL.set(false);
97 }
98 return 0;
99 }
100 };
101 testRollback(updateForEmptyTable, 1, null);
102 testRollback(updateForEmptyTable, 2, null);
103
104 final Append preAppend = new Append(ROW);
105 preAppend.add(FAMILY, QUALIFIER, VALUE);
106 Cell initCell = preAppend.getCellList(FAMILY).get(0);
107 Updater updateForNonEmptyTable = new Updater() {
108 @Override
109 public int updateData(Table table, byte[] family) throws IOException {
110 table.append(preAppend);
111 try {
112 Append append = new Append(ROW);
113 append.add(FAMILY, QUALIFIER, VALUE);
114 append.add(FAMILY, QUALIFIER_V2, VALUE);
115 FailedHLog.SHOULD_FAIL.set(true);
116 table.append(append);
117 Assert.fail("It should fail because the WAL sync is failed");
118 } catch (IOException e) {
119 } finally {
120 FailedHLog.SHOULD_FAIL.set(false);
121 }
122 return 1;
123 }
124 };
125 testRollback(updateForNonEmptyTable, 1, initCell);
126 testRollback(updateForNonEmptyTable, 2, initCell);
127 }
128
129 @Test
130 public void testIncrementRollback() throws IOException {
131 Updater updateForEmptyTable = new Updater() {
132 @Override
133 public int updateData(Table table, byte[] family) {
134 try {
135 Increment inc = new Increment(ROW);
136 inc.addColumn(FAMILY, QUALIFIER, 1);
137 inc.addColumn(FAMILY, QUALIFIER_V2, 2);
138 FailedHLog.SHOULD_FAIL.set(true);
139 table.increment(inc);
140 } catch (IOException e) {
141
142 } finally {
143 FailedHLog.SHOULD_FAIL.set(false);
144 }
145 return 0;
146 }
147 };
148 testRollback(updateForEmptyTable, 1, null);
149 testRollback(updateForEmptyTable, 2, null);
150
151 final Increment preIncrement = new Increment(ROW);
152 preIncrement.addColumn(FAMILY, QUALIFIER, 1);
153 Cell initCell = preIncrement.getCellList(FAMILY).get(0);
154 Updater updateForNonEmptyTable = new Updater() {
155 @Override
156 public int updateData(Table table, byte[] family) throws IOException {
157 table.increment(preIncrement);
158 try {
159 Increment inc = new Increment(ROW);
160 inc.addColumn(FAMILY, QUALIFIER, 1);
161 inc.addColumn(FAMILY, QUALIFIER_V2, 2);
162 FailedHLog.SHOULD_FAIL.set(true);
163 table.increment(inc);
164 Assert.fail("It should fail because the WAL sync is failed");
165 } catch (IOException e) {
166 } finally {
167 FailedHLog.SHOULD_FAIL.set(false);
168 }
169 return 1;
170 }
171 };
172 testRollback(updateForNonEmptyTable, 1, initCell);
173 testRollback(updateForNonEmptyTable, 2, initCell);
174 }
175
176 @Test
177 public void testPutRollback() throws IOException {
178 Updater updateForEmptyTable = new Updater() {
179 @Override
180 public int updateData(Table table, byte[] family) {
181 try {
182 Put put = new Put(ROW);
183 put.addColumn(FAMILY, QUALIFIER, VALUE);
184 FailedHLog.SHOULD_FAIL.set(true);
185 table.put(put);
186 Assert.fail("It should fail because the WAL sync is failed");
187 } catch (IOException e) {
188 } finally {
189 FailedHLog.SHOULD_FAIL.set(false);
190 }
191 return 0;
192 }
193 };
194 testRollback(updateForEmptyTable, 1, null);
195 testRollback(updateForEmptyTable, 2, null);
196
197 final Put prePut = new Put(ROW);
198 prePut.addColumn(FAMILY, QUALIFIER, Bytes.toBytes("aaaaaaaaaaaaaaaaaaaaaa"));
199 Cell preCell = prePut.getCellList(FAMILY).get(0);
200 Updater updateForNonEmptyTable = new Updater() {
201 @Override
202 public int updateData(Table table, byte[] family) throws IOException {
203 table.put(prePut);
204 try {
205 Put put = new Put(ROW);
206 put.addColumn(FAMILY, QUALIFIER, VALUE);
207 FailedHLog.SHOULD_FAIL.set(true);
208 table.put(put);
209 Assert.fail("It should fail because the WAL sync is failed");
210 } catch (IOException e) {
211 } finally {
212 FailedHLog.SHOULD_FAIL.set(false);
213 }
214 return 1;
215 }
216 };
217 testRollback(updateForNonEmptyTable, 1, preCell);
218 testRollback(updateForNonEmptyTable, 2, preCell);
219 }
220
221 private void testRollback(Updater updater, int versions, Cell initCell) throws IOException {
222 TableName tableName = TableName.valueOf(this.name.getMethodName());
223 HTableDescriptor desc = new HTableDescriptor(tableName);
224 HColumnDescriptor col = new HColumnDescriptor(FAMILY);
225 col.setMaxVersions(versions);
226 desc.addFamily(col);
227 TEST_UTIL.getHBaseAdmin().createTable(desc);
228 int expected;
229 List<Cell> cells;
230 try (Connection conn = ConnectionFactory.createConnection(TEST_UTIL.getConfiguration());
231 Table table = conn.getTable(tableName)) {
232 expected = updater.updateData(table, FAMILY);
233 cells = getAllCells(table);
234 }
235 TEST_UTIL.getHBaseAdmin().disableTable(tableName);
236 TEST_UTIL.getHBaseAdmin().deleteTable(tableName);
237 assertEquals(expected, cells.size());
238 if (initCell != null && cells.isEmpty()) {
239 Cell cell = cells.get(0);
240 assertTrue("row isn't matched", CellUtil.matchingRow(initCell, cell));
241 assertTrue("column isn't matched", CellUtil.matchingColumn(initCell, cell));
242 assertTrue("qualifier isn't matched", CellUtil.matchingQualifier(initCell, cell));
243 assertTrue("value isn't matched", CellUtil.matchingValue(initCell, cell));
244 }
245 }
246
247 interface Updater {
248 int updateData(Table table, byte[] family) throws IOException;
249 }
250
251 private static List<Cell> getAllCells(Table table) throws IOException {
252 List<Cell> cells = new ArrayList<>();
253 try (ResultScanner scanner = table.getScanner(new Scan())) {
254 for (Result r : scanner) {
255 cells.addAll(r.listCells());
256 }
257 return cells;
258 }
259 }
260
261 public static class FailedDefaultWALProvider extends DefaultWALProvider {
262 @Override
263 public WAL getWAL(final byte[] identifier, byte[] namespace) throws IOException {
264 WAL wal = super.getWAL(identifier, namespace);
265 return new FailedHLog(wal);
266 }
267 }
268
269 public static class FailedHLog implements WAL {
270 private static final AtomicBoolean SHOULD_FAIL = new AtomicBoolean(false);
271 private final WAL delegation;
272
273 FailedHLog(final WAL delegation) {
274 this.delegation = delegation;
275 }
276
277 @Override
278 public void registerWALActionsListener(WALActionsListener listener) {
279 delegation.registerWALActionsListener(listener);
280 }
281
282 @Override
283 public boolean unregisterWALActionsListener(WALActionsListener listener) {
284 return delegation.unregisterWALActionsListener(listener);
285 }
286
287 @Override
288 public byte[][] rollWriter() throws FailedLogCloseException, IOException {
289 return delegation.rollWriter();
290 }
291
292 @Override
293 public byte[][] rollWriter(boolean force) throws FailedLogCloseException, IOException {
294 return delegation.rollWriter(force);
295 }
296
297 @Override
298 public void shutdown() throws IOException {
299 delegation.shutdown();
300 }
301
302 @Override
303 public void close() throws IOException {
304 delegation.close();
305 }
306
307 @Override
308 public long append(HTableDescriptor htd, HRegionInfo info, WALKey key, WALEdit edits,
309 boolean inMemstore) throws IOException {
310 return delegation.append(htd, info, key, edits, inMemstore);
311 }
312
313 @Override
314 public void sync() throws IOException {
315 delegation.sync();
316 }
317
318 @Override
319 public void sync(long txid) throws IOException {
320 if (SHOULD_FAIL.get()) {
321 throw new IOException("[TESTING] we need the failure!!!");
322 }
323 delegation.sync(txid);
324 }
325
326 @Override
327 public Long startCacheFlush(byte[] encodedRegionName, Set<byte[]> families) {
328 return delegation.startCacheFlush(encodedRegionName, families);
329 }
330
331 @Override
332 public void completeCacheFlush(byte[] encodedRegionName) {
333 delegation.completeCacheFlush(encodedRegionName);
334 }
335
336 @Override
337 public void abortCacheFlush(byte[] encodedRegionName) {
338 delegation.abortCacheFlush(encodedRegionName);
339 }
340
341 @Override
342 public WALCoprocessorHost getCoprocessorHost() {
343 return delegation.getCoprocessorHost();
344 }
345
346 @Override
347 public long getEarliestMemstoreSeqNum(byte[] encodedRegionName) {
348 return delegation.getEarliestMemstoreSeqNum(encodedRegionName);
349 }
350
351 @Override
352 public long getEarliestMemstoreSeqNum(byte[] encodedRegionName, byte[] familyName) {
353 return delegation.getEarliestMemstoreSeqNum(encodedRegionName, familyName);
354 }
355
356 }
357 }