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.wal;
019
020import static org.apache.hadoop.hbase.regionserver.wal.AbstractTestWALReplay.addRegionEdits;
021import static org.apache.hadoop.hbase.wal.WALSplitter.WAL_SPLIT_TO_HFILE;
022import static org.junit.Assert.assertEquals;
023import static org.junit.Assert.assertNotNull;
024import static org.junit.Assert.assertTrue;
025import static org.junit.Assert.fail;
026import static org.mockito.Mockito.when;
027import java.io.IOException;
028import java.security.PrivilegedExceptionAction;
029import java.util.ArrayList;
030import java.util.Arrays;
031import java.util.HashMap;
032import java.util.List;
033import java.util.Map;
034import org.apache.hadoop.conf.Configuration;
035import org.apache.hadoop.fs.FSDataInputStream;
036import org.apache.hadoop.fs.FSDataOutputStream;
037import org.apache.hadoop.fs.FileStatus;
038import org.apache.hadoop.fs.FileSystem;
039import org.apache.hadoop.fs.Path;
040import org.apache.hadoop.hbase.Cell;
041import org.apache.hadoop.hbase.HBaseClassTestRule;
042import org.apache.hadoop.hbase.HBaseConfiguration;
043import org.apache.hadoop.hbase.HBaseTestingUtil;
044import org.apache.hadoop.hbase.HConstants;
045import org.apache.hadoop.hbase.ServerName;
046import org.apache.hadoop.hbase.TableName;
047import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
048import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
049import org.apache.hadoop.hbase.client.Get;
050import org.apache.hadoop.hbase.client.Put;
051import org.apache.hadoop.hbase.client.RegionInfo;
052import org.apache.hadoop.hbase.client.RegionInfoBuilder;
053import org.apache.hadoop.hbase.client.Result;
054import org.apache.hadoop.hbase.client.Scan;
055import org.apache.hadoop.hbase.client.TableDescriptor;
056import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
057import org.apache.hadoop.hbase.io.hfile.CorruptHFileException;
058import org.apache.hadoop.hbase.regionserver.DefaultStoreEngine;
059import org.apache.hadoop.hbase.regionserver.HRegion;
060import org.apache.hadoop.hbase.regionserver.RegionScanner;
061import org.apache.hadoop.hbase.regionserver.RegionServerServices;
062import org.apache.hadoop.hbase.regionserver.wal.AbstractTestWALReplay;
063import org.apache.hadoop.hbase.regionserver.wal.FSHLog;
064import org.apache.hadoop.hbase.security.User;
065import org.apache.hadoop.hbase.testclassification.MediumTests;
066import org.apache.hadoop.hbase.testclassification.RegionServerTests;
067import org.apache.hadoop.hbase.util.Bytes;
068import org.apache.hadoop.hbase.util.CommonFSUtils;
069import org.apache.hadoop.hbase.util.EnvironmentEdge;
070import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
071import org.apache.hadoop.hbase.util.FSTableDescriptors;
072import org.apache.hadoop.hbase.util.Pair;
073import org.junit.After;
074import org.junit.AfterClass;
075import org.junit.Before;
076import org.junit.BeforeClass;
077import org.junit.ClassRule;
078import org.junit.Rule;
079import org.junit.Test;
080import org.junit.experimental.categories.Category;
081import org.junit.rules.TestName;
082import org.mockito.Mockito;
083import org.slf4j.Logger;
084import org.slf4j.LoggerFactory;
085
086@Category({ RegionServerTests.class, MediumTests.class })
087public class TestWALSplitToHFile {
088  @ClassRule
089  public static final HBaseClassTestRule CLASS_RULE =
090      HBaseClassTestRule.forClass(TestWALSplitToHFile.class);
091
092  private static final Logger LOG = LoggerFactory.getLogger(AbstractTestWALReplay.class);
093  static final HBaseTestingUtil UTIL = new HBaseTestingUtil();
094  private final EnvironmentEdge ee = EnvironmentEdgeManager.getDelegate();
095  private Path rootDir = null;
096  private String logName;
097  private Path oldLogDir;
098  private Path logDir;
099  private FileSystem fs;
100  private Configuration conf;
101  private WALFactory wals;
102
103  private static final byte[] ROW = Bytes.toBytes("row");
104  private static final byte[] QUALIFIER = Bytes.toBytes("q");
105  private static final byte[] VALUE1 = Bytes.toBytes("value1");
106  private static final byte[] VALUE2 = Bytes.toBytes("value2");
107  private static final int countPerFamily = 10;
108
109  @Rule
110  public final TestName TEST_NAME = new TestName();
111
112  @BeforeClass
113  public static void setUpBeforeClass() throws Exception {
114    Configuration conf = UTIL.getConfiguration();
115    conf.setBoolean(WAL_SPLIT_TO_HFILE, true);
116    UTIL.startMiniCluster(3);
117    Path hbaseRootDir = UTIL.getDFSCluster().getFileSystem().makeQualified(new Path("/hbase"));
118    LOG.info("hbase.rootdir=" + hbaseRootDir);
119    CommonFSUtils.setRootDir(conf, hbaseRootDir);
120  }
121
122  @AfterClass
123  public static void tearDownAfterClass() throws Exception {
124    UTIL.shutdownMiniCluster();
125  }
126
127  @Before
128  public void setUp() throws Exception {
129    this.conf = HBaseConfiguration.create(UTIL.getConfiguration());
130    this.conf.setBoolean(HConstants.HREGION_EDITS_REPLAY_SKIP_ERRORS, false);
131    this.fs = UTIL.getDFSCluster().getFileSystem();
132    this.rootDir = CommonFSUtils.getRootDir(this.conf);
133    this.oldLogDir = new Path(this.rootDir, HConstants.HREGION_OLDLOGDIR_NAME);
134    String serverName =
135      ServerName.valueOf(TEST_NAME.getMethodName() + "-manual", 16010,
136        EnvironmentEdgeManager.currentTime())
137          .toString();
138    this.logName = AbstractFSWALProvider.getWALDirectoryName(serverName);
139    this.logDir = new Path(this.rootDir, logName);
140    if (UTIL.getDFSCluster().getFileSystem().exists(this.rootDir)) {
141      UTIL.getDFSCluster().getFileSystem().delete(this.rootDir, true);
142    }
143    this.wals = new WALFactory(conf, TEST_NAME.getMethodName());
144  }
145
146  @After
147  public void tearDown() throws Exception {
148    this.wals.close();
149    UTIL.getDFSCluster().getFileSystem().delete(this.rootDir, true);
150  }
151
152  /*
153   * @param p Directory to cleanup
154   */
155  private void deleteDir(final Path p) throws IOException {
156    if (this.fs.exists(p)) {
157      if (!this.fs.delete(p, true)) {
158        throw new IOException("Failed remove of " + p);
159      }
160    }
161  }
162
163  private TableDescriptor createBasic3FamilyTD(final TableName tableName) throws IOException {
164    TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName);
165    builder.setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("a")).build());
166    builder.setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("b")).build());
167    builder.setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("c")).build());
168    TableDescriptor td = builder.build();
169    UTIL.getAdmin().createTable(td);
170    return td;
171  }
172
173  private WAL createWAL(Configuration c, Path hbaseRootDir, String logName) throws IOException {
174    FSHLog wal = new FSHLog(FileSystem.get(c), hbaseRootDir, logName, c);
175    wal.init();
176    return wal;
177  }
178
179  private WAL createWAL(FileSystem fs, Path hbaseRootDir, String logName) throws IOException {
180    FSHLog wal = new FSHLog(fs, hbaseRootDir, logName, this.conf);
181    wal.init();
182    return wal;
183  }
184
185  private Pair<TableDescriptor, RegionInfo> setupTableAndRegion() throws IOException {
186    final TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());
187    final TableDescriptor td = createBasic3FamilyTD(tableName);
188    final RegionInfo ri = RegionInfoBuilder.newBuilder(tableName).build();
189    final Path tableDir = CommonFSUtils.getTableDir(this.rootDir, tableName);
190    deleteDir(tableDir);
191    FSTableDescriptors.createTableDescriptorForTableDirectory(fs, tableDir, td, false);
192    HRegion region = HBaseTestingUtil.createRegionAndWAL(ri, rootDir, this.conf, td);
193    HBaseTestingUtil.closeRegionAndWAL(region);
194    return new Pair<>(td, ri);
195  }
196
197  private void writeData(TableDescriptor td, HRegion region) throws IOException {
198    final long timestamp = this.ee.currentTime();
199    for (ColumnFamilyDescriptor cfd : td.getColumnFamilies()) {
200      region.put(new Put(ROW).addColumn(cfd.getName(), QUALIFIER, timestamp, VALUE1));
201    }
202  }
203
204  @Test
205  public void testDifferentRootDirAndWALRootDir() throws Exception {
206    // Change wal root dir and reset the configuration
207    Path walRootDir = UTIL.createWALRootDir();
208    this.conf = HBaseConfiguration.create(UTIL.getConfiguration());
209
210    FileSystem walFs = CommonFSUtils.getWALFileSystem(this.conf);
211    this.oldLogDir = new Path(walRootDir, HConstants.HREGION_OLDLOGDIR_NAME);
212    String serverName =
213      ServerName.valueOf(TEST_NAME.getMethodName() + "-manual", 16010,
214        EnvironmentEdgeManager.currentTime())
215          .toString();
216    this.logName = AbstractFSWALProvider.getWALDirectoryName(serverName);
217    this.logDir = new Path(walRootDir, logName);
218    this.wals = new WALFactory(conf, TEST_NAME.getMethodName());
219
220    Pair<TableDescriptor, RegionInfo> pair = setupTableAndRegion();
221    TableDescriptor td = pair.getFirst();
222    RegionInfo ri = pair.getSecond();
223
224    WAL wal = createWAL(walFs, walRootDir, logName);
225    HRegion region = HRegion.openHRegion(this.conf, this.fs, rootDir, ri, td, wal);
226    writeData(td, region);
227
228    // Now close the region without flush
229    region.close(true);
230    wal.shutdown();
231    // split the log
232    WALSplitter.split(walRootDir, logDir, oldLogDir, FileSystem.get(this.conf), this.conf, wals);
233
234    WAL wal2 = createWAL(walFs, walRootDir, logName);
235    HRegion region2 = HRegion.openHRegion(this.conf, this.fs, rootDir, ri, td, wal2);
236    Result result2 = region2.get(new Get(ROW));
237    assertEquals(td.getColumnFamilies().length, result2.size());
238    for (ColumnFamilyDescriptor cfd : td.getColumnFamilies()) {
239      assertTrue(Bytes.equals(VALUE1, result2.getValue(cfd.getName(), QUALIFIER)));
240    }
241  }
242
243  @Test
244  public void testCorruptRecoveredHFile() throws Exception {
245    Pair<TableDescriptor, RegionInfo> pair = setupTableAndRegion();
246    TableDescriptor td = pair.getFirst();
247    RegionInfo ri = pair.getSecond();
248
249    WAL wal = createWAL(this.conf, rootDir, logName);
250    HRegion region = HRegion.openHRegion(this.conf, this.fs, rootDir, ri, td, wal);
251    writeData(td, region);
252
253    // Now close the region without flush
254    region.close(true);
255    wal.shutdown();
256    // split the log
257    WALSplitter.split(rootDir, logDir, oldLogDir, FileSystem.get(this.conf), this.conf, wals);
258
259    // Write a corrupt recovered hfile
260    Path regionDir =
261        new Path(CommonFSUtils.getTableDir(rootDir, td.getTableName()), ri.getEncodedName());
262    for (ColumnFamilyDescriptor cfd : td.getColumnFamilies()) {
263      FileStatus[] files =
264          WALSplitUtil.getRecoveredHFiles(this.fs, regionDir, cfd.getNameAsString());
265      assertNotNull(files);
266      assertTrue(files.length > 0);
267      writeCorruptRecoveredHFile(files[0].getPath());
268    }
269
270    // Failed to reopen the region
271    WAL wal2 = createWAL(this.conf, rootDir, logName);
272    try {
273      HRegion.openHRegion(this.conf, this.fs, rootDir, ri, td, wal2);
274      fail("Should fail to open region");
275    } catch (CorruptHFileException che) {
276      // Expected
277    }
278
279    // Set skip errors to true and reopen the region
280    this.conf.setBoolean(HConstants.HREGION_EDITS_REPLAY_SKIP_ERRORS, true);
281    HRegion region2 = HRegion.openHRegion(this.conf, this.fs, rootDir, ri, td, wal2);
282    Result result2 = region2.get(new Get(ROW));
283    assertEquals(td.getColumnFamilies().length, result2.size());
284    for (ColumnFamilyDescriptor cfd : td.getColumnFamilies()) {
285      assertTrue(Bytes.equals(VALUE1, result2.getValue(cfd.getName(), QUALIFIER)));
286      // Assert the corrupt file was skipped and still exist
287      FileStatus[] files =
288          WALSplitUtil.getRecoveredHFiles(this.fs, regionDir, cfd.getNameAsString());
289      assertNotNull(files);
290      assertEquals(1, files.length);
291      assertTrue(files[0].getPath().getName().contains("corrupt"));
292    }
293  }
294
295  @Test
296  public void testPutWithSameTimestamp() throws Exception {
297    Pair<TableDescriptor, RegionInfo> pair = setupTableAndRegion();
298    TableDescriptor td = pair.getFirst();
299    RegionInfo ri = pair.getSecond();
300
301    WAL wal = createWAL(this.conf, rootDir, logName);
302    HRegion region = HRegion.openHRegion(this.conf, this.fs, rootDir, ri, td, wal);
303    final long timestamp = this.ee.currentTime();
304    // Write data and flush
305    for (ColumnFamilyDescriptor cfd : td.getColumnFamilies()) {
306      region.put(new Put(ROW).addColumn(cfd.getName(), QUALIFIER, timestamp, VALUE1));
307    }
308    region.flush(true);
309
310    // Write data with same timestamp and do not flush
311    for (ColumnFamilyDescriptor cfd : td.getColumnFamilies()) {
312      region.put(new Put(ROW).addColumn(cfd.getName(), QUALIFIER, timestamp, VALUE2));
313    }
314    // Now close the region without flush
315    region.close(true);
316    wal.shutdown();
317    // split the log
318    WALSplitter.split(rootDir, logDir, oldLogDir, FileSystem.get(this.conf), this.conf, wals);
319
320    // reopen the region
321    WAL wal2 = createWAL(this.conf, rootDir, logName);
322    HRegion region2 = HRegion.openHRegion(conf, this.fs, rootDir, ri, td, wal2);
323    Result result2 = region2.get(new Get(ROW));
324    assertEquals(td.getColumnFamilies().length, result2.size());
325    for (ColumnFamilyDescriptor cfd : td.getColumnFamilies()) {
326      assertTrue(Bytes.equals(VALUE2, result2.getValue(cfd.getName(), QUALIFIER)));
327    }
328  }
329
330  @Test
331  public void testRecoverSequenceId() throws Exception {
332    Pair<TableDescriptor, RegionInfo> pair = setupTableAndRegion();
333    TableDescriptor td = pair.getFirst();
334    RegionInfo ri = pair.getSecond();
335
336    WAL wal = createWAL(this.conf, rootDir, logName);
337    HRegion region = HRegion.openHRegion(this.conf, this.fs, rootDir, ri, td, wal);
338    Map<Integer, Map<String, Long>> seqIdMap = new HashMap<>();
339    // Write data and do not flush
340    for (int i = 0; i < countPerFamily; i++) {
341      for (ColumnFamilyDescriptor cfd : td.getColumnFamilies()) {
342        region.put(new Put(Bytes.toBytes(i)).addColumn(cfd.getName(), QUALIFIER, VALUE1));
343        Result result = region.get(new Get(Bytes.toBytes(i)).addFamily(cfd.getName()));
344        assertTrue(Bytes.equals(VALUE1, result.getValue(cfd.getName(), QUALIFIER)));
345        List<Cell> cells = result.listCells();
346        assertEquals(1, cells.size());
347        seqIdMap.computeIfAbsent(i, r -> new HashMap<>()).put(cfd.getNameAsString(),
348          cells.get(0).getSequenceId());
349      }
350    }
351
352    // Now close the region without flush
353    region.close(true);
354    wal.shutdown();
355    // split the log
356    WALSplitter.split(rootDir, logDir, oldLogDir, FileSystem.get(this.conf), this.conf, wals);
357
358    // reopen the region
359    WAL wal2 = createWAL(this.conf, rootDir, logName);
360    HRegion region2 = HRegion.openHRegion(conf, this.fs, rootDir, ri, td, wal2);
361    // assert the seqid was recovered
362    for (int i = 0; i < countPerFamily; i++) {
363      for (ColumnFamilyDescriptor cfd : td.getColumnFamilies()) {
364        Result result = region2.get(new Get(Bytes.toBytes(i)).addFamily(cfd.getName()));
365        assertTrue(Bytes.equals(VALUE1, result.getValue(cfd.getName(), QUALIFIER)));
366        List<Cell> cells = result.listCells();
367        assertEquals(1, cells.size());
368        assertEquals((long) seqIdMap.get(i).get(cfd.getNameAsString()),
369          cells.get(0).getSequenceId());
370      }
371    }
372  }
373
374  /**
375   * Test writing edits into an HRegion, closing it, splitting logs, opening
376   * Region again.  Verify seqids.
377   */
378  @Test
379  public void testWrittenViaHRegion()
380      throws IOException, SecurityException, IllegalArgumentException, InterruptedException {
381    Pair<TableDescriptor, RegionInfo> pair = setupTableAndRegion();
382    TableDescriptor td = pair.getFirst();
383    RegionInfo ri = pair.getSecond();
384
385    // Write countPerFamily edits into the three families.  Do a flush on one
386    // of the families during the load of edits so its seqid is not same as
387    // others to test we do right thing when different seqids.
388    WAL wal = createWAL(this.conf, rootDir, logName);
389    HRegion region = HRegion.openHRegion(this.conf, this.fs, rootDir, ri, td, wal);
390    long seqid = region.getOpenSeqNum();
391    boolean first = true;
392    for (ColumnFamilyDescriptor cfd : td.getColumnFamilies()) {
393      addRegionEdits(ROW, cfd.getName(), countPerFamily, this.ee, region, "x");
394      if (first) {
395        // If first, so we have at least one family w/ different seqid to rest.
396        region.flush(true);
397        first = false;
398      }
399    }
400    // Now assert edits made it in.
401    final Get g = new Get(ROW);
402    Result result = region.get(g);
403    assertEquals(countPerFamily * td.getColumnFamilies().length, result.size());
404    // Now close the region (without flush), split the log, reopen the region and assert that
405    // replay of log has the correct effect, that our seqids are calculated correctly so
406    // all edits in logs are seen as 'stale'/old.
407    region.close(true);
408    wal.shutdown();
409    try {
410      WALSplitter.split(rootDir, logDir, oldLogDir, FileSystem.get(this.conf), this.conf, wals);
411    } catch (Exception e) {
412      LOG.debug("Got exception", e);
413    }
414
415    WAL wal2 = createWAL(this.conf, rootDir, logName);
416    HRegion region2 = HRegion.openHRegion(conf, this.fs, rootDir, ri, td, wal2);
417    long seqid2 = region2.getOpenSeqNum();
418    assertTrue(seqid + result.size() < seqid2);
419    final Result result1b = region2.get(g);
420    assertEquals(result.size(), result1b.size());
421
422    // Next test.  Add more edits, then 'crash' this region by stealing its wal
423    // out from under it and assert that replay of the log adds the edits back
424    // correctly when region is opened again.
425    for (ColumnFamilyDescriptor hcd : td.getColumnFamilies()) {
426      addRegionEdits(ROW, hcd.getName(), countPerFamily, this.ee, region2, "y");
427    }
428    // Get count of edits.
429    final Result result2 = region2.get(g);
430    assertEquals(2 * result.size(), result2.size());
431    wal2.sync();
432    final Configuration newConf = HBaseConfiguration.create(this.conf);
433    User user = HBaseTestingUtil.getDifferentUser(newConf, td.getTableName().getNameAsString());
434    user.runAs(new PrivilegedExceptionAction<Object>() {
435      @Override
436      public Object run() throws Exception {
437        WALSplitter.split(rootDir, logDir, oldLogDir, FileSystem.get(conf), conf, wals);
438        FileSystem newFS = FileSystem.get(newConf);
439        // Make a new wal for new region open.
440        WAL wal3 = createWAL(newConf, rootDir, logName);
441        Path tableDir = CommonFSUtils.getTableDir(rootDir, td.getTableName());
442        HRegion region3 = new HRegion(tableDir, wal3, newFS, newConf, ri, td, null);
443        long seqid3 = region3.initialize();
444        Result result3 = region3.get(g);
445        // Assert that count of cells is same as before crash.
446        assertEquals(result2.size(), result3.size());
447
448        // I can't close wal1.  Its been appropriated when we split.
449        region3.close();
450        wal3.close();
451        return null;
452      }
453    });
454  }
455
456  /**
457   * Test that we recover correctly when there is a failure in between the
458   * flushes. i.e. Some stores got flushed but others did not.
459   * Unfortunately, there is no easy hook to flush at a store level. The way
460   * we get around this is by flushing at the region level, and then deleting
461   * the recently flushed store file for one of the Stores. This would put us
462   * back in the situation where all but that store got flushed and the region
463   * died.
464   * We restart Region again, and verify that the edits were replayed.
465   */
466  @Test
467  public void testAfterPartialFlush()
468      throws IOException, SecurityException, IllegalArgumentException {
469    Pair<TableDescriptor, RegionInfo> pair = setupTableAndRegion();
470    TableDescriptor td = pair.getFirst();
471    RegionInfo ri = pair.getSecond();
472
473    // Write countPerFamily edits into the three families.  Do a flush on one
474    // of the families during the load of edits so its seqid is not same as
475    // others to test we do right thing when different seqids.
476    WAL wal = createWAL(this.conf, rootDir, logName);
477    HRegion region = HRegion.openHRegion(this.conf, this.fs, rootDir, ri, td, wal);
478    long seqid = region.getOpenSeqNum();
479    for (ColumnFamilyDescriptor cfd : td.getColumnFamilies()) {
480      addRegionEdits(ROW, cfd.getName(), countPerFamily, this.ee, region, "x");
481    }
482
483    // Now assert edits made it in.
484    final Get g = new Get(ROW);
485    Result result = region.get(g);
486    assertEquals(countPerFamily * td.getColumnFamilies().length, result.size());
487
488    // Let us flush the region
489    region.flush(true);
490    region.close(true);
491    wal.shutdown();
492
493    // delete the store files in the second column family to simulate a failure
494    // in between the flushcache();
495    // we have 3 families. killing the middle one ensures that taking the maximum
496    // will make us fail.
497    int cf_count = 0;
498    for (ColumnFamilyDescriptor cfd : td.getColumnFamilies()) {
499      cf_count++;
500      if (cf_count == 2) {
501        region.getRegionFileSystem().deleteFamily(cfd.getNameAsString());
502      }
503    }
504
505    // Let us try to split and recover
506    WALSplitter.split(rootDir, logDir, oldLogDir, FileSystem.get(this.conf), this.conf, wals);
507    WAL wal2 = createWAL(this.conf, rootDir, logName);
508    HRegion region2 = HRegion.openHRegion(this.conf, this.fs, rootDir, ri, td, wal2);
509    long seqid2 = region2.getOpenSeqNum();
510    assertTrue(seqid + result.size() < seqid2);
511
512    final Result result1b = region2.get(g);
513    assertEquals(result.size(), result1b.size());
514  }
515
516  /**
517   * Test that we could recover the data correctly after aborting flush. In the
518   * test, first we abort flush after writing some data, then writing more data
519   * and flush again, at last verify the data.
520   */
521  @Test
522  public void testAfterAbortingFlush() throws IOException {
523    Pair<TableDescriptor, RegionInfo> pair = setupTableAndRegion();
524    TableDescriptor td = pair.getFirst();
525    RegionInfo ri = pair.getSecond();
526
527    // Write countPerFamily edits into the three families. Do a flush on one
528    // of the families during the load of edits so its seqid is not same as
529    // others to test we do right thing when different seqids.
530    WAL wal = createWAL(this.conf, rootDir, logName);
531    RegionServerServices rsServices = Mockito.mock(RegionServerServices.class);
532    Mockito.doReturn(false).when(rsServices).isAborted();
533    when(rsServices.getServerName()).thenReturn(ServerName.valueOf("foo", 10, 10));
534    when(rsServices.getConfiguration()).thenReturn(conf);
535    Configuration customConf = new Configuration(this.conf);
536    customConf.set(DefaultStoreEngine.DEFAULT_STORE_FLUSHER_CLASS_KEY,
537        AbstractTestWALReplay.CustomStoreFlusher.class.getName());
538    HRegion region = HRegion.openHRegion(this.rootDir, ri, td, wal, customConf, rsServices, null);
539    int writtenRowCount = 10;
540    List<ColumnFamilyDescriptor> families = Arrays.asList(td.getColumnFamilies());
541    for (int i = 0; i < writtenRowCount; i++) {
542      Put put = new Put(Bytes.toBytes(td.getTableName() + Integer.toString(i)));
543      put.addColumn(families.get(i % families.size()).getName(), Bytes.toBytes("q"),
544          Bytes.toBytes("val"));
545      region.put(put);
546    }
547
548    // Now assert edits made it in.
549    RegionScanner scanner = region.getScanner(new Scan());
550    assertEquals(writtenRowCount, getScannedCount(scanner));
551
552    // Let us flush the region
553    AbstractTestWALReplay.CustomStoreFlusher.throwExceptionWhenFlushing.set(true);
554    try {
555      region.flush(true);
556      fail("Injected exception hasn't been thrown");
557    } catch (IOException e) {
558      LOG.info("Expected simulated exception when flushing region, {}", e.getMessage());
559      // simulated to abort server
560      Mockito.doReturn(true).when(rsServices).isAborted();
561      region.setClosing(false); // region normally does not accept writes after
562      // DroppedSnapshotException. We mock around it for this test.
563    }
564    // writing more data
565    int moreRow = 10;
566    for (int i = writtenRowCount; i < writtenRowCount + moreRow; i++) {
567      Put put = new Put(Bytes.toBytes(td.getTableName() + Integer.toString(i)));
568      put.addColumn(families.get(i % families.size()).getName(), Bytes.toBytes("q"),
569          Bytes.toBytes("val"));
570      region.put(put);
571    }
572    writtenRowCount += moreRow;
573    // call flush again
574    AbstractTestWALReplay.CustomStoreFlusher.throwExceptionWhenFlushing.set(false);
575    try {
576      region.flush(true);
577    } catch (IOException t) {
578      LOG.info(
579          "Expected exception when flushing region because server is stopped," + t.getMessage());
580    }
581
582    region.close(true);
583    wal.shutdown();
584
585    // Let us try to split and recover
586    WALSplitter.split(rootDir, logDir, oldLogDir, FileSystem.get(this.conf), this.conf, wals);
587    WAL wal2 = createWAL(this.conf, rootDir, logName);
588    Mockito.doReturn(false).when(rsServices).isAborted();
589    HRegion region2 = HRegion.openHRegion(this.rootDir, ri, td, wal2, this.conf, rsServices, null);
590    scanner = region2.getScanner(new Scan());
591    assertEquals(writtenRowCount, getScannedCount(scanner));
592  }
593
594  private int getScannedCount(RegionScanner scanner) throws IOException {
595    int scannedCount = 0;
596    List<Cell> results = new ArrayList<>();
597    while (true) {
598      boolean existMore = scanner.next(results);
599      if (!results.isEmpty()) {
600        scannedCount++;
601      }
602      if (!existMore) {
603        break;
604      }
605      results.clear();
606    }
607    return scannedCount;
608  }
609
610  private void writeCorruptRecoveredHFile(Path recoveredHFile) throws Exception {
611    // Read the recovered hfile
612    int fileSize = (int) fs.listStatus(recoveredHFile)[0].getLen();
613    FSDataInputStream in = fs.open(recoveredHFile);
614    byte[] fileContent = new byte[fileSize];
615    in.readFully(0, fileContent, 0, fileSize);
616    in.close();
617
618    // Write a corrupt hfile by append garbage
619    Path path = new Path(recoveredHFile.getParent(), recoveredHFile.getName() + ".corrupt");
620    FSDataOutputStream out;
621    out = fs.create(path);
622    out.write(fileContent);
623    out.write(Bytes.toBytes("-----"));
624    out.close();
625  }
626}