hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactionArchiveConcurrentClose.java

   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 package org.apache.hadoop.hbase.regionserver;
  19
  20 import static org.junit.Assert.assertEquals;
  21 import static org.junit.Assert.assertFalse;
  22 import static org.mockito.Mockito.mock;
  23
  24 import java.io.IOException;
  25 import java.io.InterruptedIOException;
  26 import java.util.ArrayList;
  27 import java.util.Collection;
  28 import java.util.List;
  29 import java.util.concurrent.atomic.AtomicBoolean;
  30 import java.util.concurrent.atomic.AtomicReference;
  31 import org.apache.hadoop.conf.Configuration;
  32 import org.apache.hadoop.fs.FileSystem;
  33 import org.apache.hadoop.fs.Path;
  34 import org.apache.hadoop.hbase.HBaseClassTestRule;
  35 import org.apache.hadoop.hbase.HBaseTestingUtil;
  36 import org.apache.hadoop.hbase.Stoppable;
  37 import org.apache.hadoop.hbase.TableName;
  38 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
  39 import org.apache.hadoop.hbase.client.Put;
  40 import org.apache.hadoop.hbase.client.RegionInfo;
  41 import org.apache.hadoop.hbase.client.RegionInfoBuilder;
  42 import org.apache.hadoop.hbase.client.TableDescriptor;
  43 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
  44 import org.apache.hadoop.hbase.testclassification.RegionServerTests;
  45 import org.apache.hadoop.hbase.testclassification.SmallTests;
  46 import org.apache.hadoop.hbase.util.Bytes;
  47 import org.apache.hadoop.hbase.util.CommonFSUtils;
  48 import org.apache.hadoop.hbase.wal.WALFactory;
  49 import org.junit.After;
  50 import org.junit.Before;
  51 import org.junit.ClassRule;
  52 import org.junit.Rule;
  53 import org.junit.Test;
  54 import org.junit.experimental.categories.Category;
  55 import org.junit.rules.TestName;
  56 import org.mockito.Mockito;
  57
  58 /**
  59  * Tests a race condition between archiving of compacted files in CompactedHFilesDischarger chore
  60  * and HRegion.close();
  61  */
  62 @Category({RegionServerTests.class, SmallTests.class})
  63 public class TestCompactionArchiveConcurrentClose {
  64
  65   @ClassRule
  66   public static final HBaseClassTestRule CLASS_RULE =
  67       HBaseClassTestRule.forClass(TestCompactionArchiveConcurrentClose.class);
  68
  69   private HBaseTestingUtil testUtil;
  70
  71   private Path testDir;
  72   private AtomicBoolean archived = new AtomicBoolean();
  73
  74   @Rule
  75   public TestName name = new TestName();
  76
  77   @Before
  78   public void setup() throws Exception {
  79     testUtil = new HBaseTestingUtil();
  80     testDir = testUtil.getDataTestDir("TestStoreFileRefresherChore");
  81     CommonFSUtils.setRootDir(testUtil.getConfiguration(), testDir);
  82   }
  83
  84   @After
  85   public void tearDown() throws Exception {
  86     testUtil.cleanupTestDir();
  87   }
  88
  89   @Test
  90   public void testStoreCloseAndDischargeRunningInParallel() throws Exception {
  91     byte[] fam = Bytes.toBytes("f");
  92     byte[] col = Bytes.toBytes("c");
  93     byte[] val = Bytes.toBytes("val");
  94
  95     TableName tableName = TableName.valueOf(name.getMethodName());
  96     TableDescriptor htd = TableDescriptorBuilder.newBuilder(tableName)
  97         .setColumnFamily(ColumnFamilyDescriptorBuilder.of(fam)).build();
  98     RegionInfo info = RegionInfoBuilder.newBuilder(tableName).build();
  99     HRegion region = initHRegion(htd, info);
 100     RegionServerServices rss = mock(RegionServerServices.class);
 101     List<HRegion> regions = new ArrayList<>();
 102     regions.add(region);
 103     Mockito.doReturn(regions).when(rss).getRegions();
 104
 105     // Create the cleaner object
 106     CompactedHFilesDischarger cleaner =
 107         new CompactedHFilesDischarger(1000, (Stoppable) null, rss, false);
 108     // Add some data to the region and do some flushes
 109     int batchSize = 10;
 110     int fileCount = 10;
 111     for (int f = 0; f < fileCount; f++) {
 112       int start = f * batchSize;
 113       for (int i = start; i < start + batchSize; i++) {
 114         Put p = new Put(Bytes.toBytes("row" + i));
 115         p.addColumn(fam, col, val);
 116         region.put(p);
 117       }
 118       // flush them
 119       region.flush(true);
 120     }
 121
 122     HStore store = region.getStore(fam);
 123     assertEquals(fileCount, store.getStorefilesCount());
 124
 125     Collection<HStoreFile> storefiles = store.getStorefiles();
 126     // None of the files should be in compacted state.
 127     for (HStoreFile file : storefiles) {
 128       assertFalse(file.isCompactedAway());
 129     }
 130     // Do compaction
 131     region.compact(true);
 132
 133     // now run the cleaner with a concurrent close
 134     Thread cleanerThread = new Thread() {
 135       @Override
 136       public void run() {
 137         cleaner.chore();
 138       }
 139     };
 140     cleanerThread.start();
 141     // wait for cleaner to pause
 142     synchronized (archived) {
 143       if (!archived.get()) {
 144         archived.wait();
 145       }
 146     }
 147     final AtomicReference<Exception> closeException = new AtomicReference<>();
 148     Thread closeThread = new Thread() {
 149       @Override
 150       public void run() {
 151         // wait for the chore to complete and call close
 152         try {
 153           ((HRegion) region).close();
 154         } catch (IOException e) {
 155           closeException.set(e);
 156         }
 157       }
 158     };
 159     closeThread.start();
 160     // no error should occur after the execution of the test
 161     closeThread.join();
 162     cleanerThread.join();
 163
 164     if (closeException.get() != null) {
 165       throw closeException.get();
 166     }
 167   }
 168
 169   private HRegion initHRegion(TableDescriptor htd, RegionInfo info) throws IOException {
 170     Configuration conf = testUtil.getConfiguration();
 171     Path tableDir = CommonFSUtils.getTableDir(testDir, htd.getTableName());
 172
 173     HRegionFileSystem fs =
 174         new WaitingHRegionFileSystem(conf, tableDir.getFileSystem(conf), tableDir, info);
 175     ChunkCreator.initialize(MemStoreLAB.CHUNK_SIZE_DEFAULT, false, 0, 0,
 176       0, null, MemStoreLAB.INDEX_CHUNK_SIZE_PERCENTAGE_DEFAULT);
 177     final Configuration walConf = new Configuration(conf);
 178     CommonFSUtils.setRootDir(walConf, tableDir);
 179     final WALFactory wals = new WALFactory(walConf, "log_" + info.getEncodedName());
 180     HRegion region = new HRegion(fs, wals.getWAL(info), conf, htd, null);
 181
 182     region.initialize();
 183
 184     return region;
 185   }
 186
 187   private class WaitingHRegionFileSystem extends HRegionFileSystem {
 188
 189     public WaitingHRegionFileSystem(final Configuration conf, final FileSystem fs,
 190         final Path tableDir, final RegionInfo regionInfo) {
 191       super(conf, fs, tableDir, regionInfo);
 192     }
 193
 194     @Override
 195     public void removeStoreFiles(String familyName, Collection<HStoreFile> storeFiles)
 196         throws IOException {
 197       super.removeStoreFiles(familyName, storeFiles);
 198       archived.set(true);
 199       synchronized (archived) {
 200         archived.notifyAll();
 201       }
 202       try {
 203         // unfortunately we can't use a stronger barrier here as the fix synchronizing
 204         // the race condition will then block
 205         Thread.sleep(100);
 206       } catch (InterruptedException ie) {
 207         throw new InterruptedIOException("Interrupted waiting for latch");
 208       }
 209     }
 210   }
 211 }