1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.regionserver;
19
20 import static org.junit.Assert.assertArrayEquals;
21 import static org.junit.Assert.assertTrue;
22 import static org.mockito.Mockito.doAnswer;
23 import static org.mockito.Mockito.spy;
24
25 import java.io.IOException;
26 import java.util.Collection;
27 import java.util.Map;
28
29 import org.apache.commons.lang.mutable.MutableBoolean;
30 import org.apache.commons.logging.Log;
31 import org.apache.commons.logging.LogFactory;
32 import org.apache.hadoop.hbase.DroppedSnapshotException;
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.HTableDescriptor;
37 import org.apache.hadoop.hbase.NamespaceDescriptor;
38 import org.apache.hadoop.hbase.TableName;
39 import org.apache.hadoop.hbase.client.Connection;
40 import org.apache.hadoop.hbase.client.Get;
41 import org.apache.hadoop.hbase.client.HBaseAdmin;
42 import org.apache.hadoop.hbase.client.Put;
43 import org.apache.hadoop.hbase.client.Result;
44 import org.apache.hadoop.hbase.client.Table;
45 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
46 import org.apache.hadoop.hbase.regionserver.HRegion.PrepareFlushResult;
47 import org.apache.hadoop.hbase.regionserver.Region.FlushResult;
48 import org.apache.hadoop.hbase.testclassification.MediumTests;
49 import org.apache.hadoop.hbase.util.Bytes;
50 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
51 import org.apache.hadoop.hbase.wal.WAL;
52 import org.junit.After;
53 import org.junit.Before;
54 import org.junit.Test;
55 import org.junit.experimental.categories.Category;
56 import org.mockito.Matchers;
57 import org.mockito.invocation.InvocationOnMock;
58 import org.mockito.stubbing.Answer;
59
60
61
62
63 @Category({ MediumTests.class })
64 public class TestSplitWalDataLoss {
65
66 private static final Log LOG = LogFactory.getLog(TestSplitWalDataLoss.class);
67
68 private final HBaseTestingUtility testUtil = new HBaseTestingUtility();
69
70 private NamespaceDescriptor namespace = NamespaceDescriptor.create(getClass().getSimpleName())
71 .build();
72
73 private TableName tableName = TableName.valueOf(namespace.getName(), "dataloss");
74
75 private byte[] family = Bytes.toBytes("f");
76
77 private byte[] qualifier = Bytes.toBytes("q");
78
79 @Before
80 public void setUp() throws Exception {
81 testUtil.getConfiguration().setInt("hbase.regionserver.msginterval", 30000);
82 testUtil.getConfiguration().setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, false);
83 testUtil.startMiniCluster(2);
84 HBaseAdmin admin = testUtil.getHBaseAdmin();
85 admin.createNamespace(namespace);
86 HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName);
87 hTableDescriptor.addFamily(new HColumnDescriptor(family));
88 admin.createTable(hTableDescriptor);
89 testUtil.waitTableAvailable(tableName);
90 }
91
92 @After
93 public void tearDown() throws Exception {
94 testUtil.shutdownMiniCluster();
95 }
96
97 @Test
98 public void test() throws IOException, InterruptedException {
99 final HRegionServer rs = testUtil.getRSForFirstRegionInTable(tableName);
100 final HRegion region = (HRegion) rs.getOnlineRegions(tableName).get(0);
101 HRegion spiedRegion = spy(region);
102 final MutableBoolean flushed = new MutableBoolean(false);
103 final MutableBoolean reported = new MutableBoolean(false);
104 doAnswer(new Answer<FlushResult>() {
105 @Override
106 public FlushResult answer(InvocationOnMock invocation) throws Throwable {
107 synchronized (flushed) {
108 flushed.setValue(true);
109 flushed.notifyAll();
110 }
111 synchronized (reported) {
112 while (!reported.booleanValue()) {
113 reported.wait();
114 }
115 }
116 rs.getWAL(region.getRegionInfo()).abortCacheFlush(
117 region.getRegionInfo().getEncodedNameAsBytes());
118 throw new DroppedSnapshotException("testcase");
119 }
120 }).when(spiedRegion).internalFlushCacheAndCommit(Matchers.<WAL> any(),
121 Matchers.<MonitoredTask> any(), Matchers.<PrepareFlushResult> any(),
122 Matchers.<Collection<Store>> any());
123
124 String key = null;
125 for (Map.Entry<String, Region> entry: rs.onlineRegions.entrySet()) {
126 if (entry.getValue().getRegionInfo().getTable().equals(this.tableName)) {
127 key = entry.getKey();
128 break;
129 }
130 }
131 rs.onlineRegions.put(key, spiedRegion);
132 Connection conn = testUtil.getConnection();
133
134 try (Table table = conn.getTable(tableName)) {
135 table.put(new Put(Bytes.toBytes("row0")).addColumn(family, qualifier, Bytes.toBytes("val0")));
136 }
137 long oldestSeqIdOfStore = region.getOldestSeqIdOfStore(family);
138 LOG.info("CHANGE OLDEST " + oldestSeqIdOfStore);
139 assertTrue(oldestSeqIdOfStore > HConstants.NO_SEQNUM);
140 rs.cacheFlusher.requestFlush(spiedRegion, false);
141 synchronized (flushed) {
142 while (!flushed.booleanValue()) {
143 flushed.wait();
144 }
145 }
146 try (Table table = conn.getTable(tableName)) {
147 table.put(new Put(Bytes.toBytes("row1")).addColumn(family, qualifier, Bytes.toBytes("val1")));
148 }
149 long now = EnvironmentEdgeManager.currentTime();
150 rs.tryRegionServerReport(now - 500, now);
151 synchronized (reported) {
152 reported.setValue(true);
153 reported.notifyAll();
154 }
155 while (testUtil.getRSForFirstRegionInTable(tableName) == rs) {
156 Thread.sleep(100);
157 }
158 try (Table table = conn.getTable(tableName)) {
159 Result result = table.get(new Get(Bytes.toBytes("row0")));
160 assertArrayEquals(Bytes.toBytes("val0"), result.getValue(family, qualifier));
161 }
162 }
163 }