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.regionserver;
019
020import static org.junit.jupiter.api.Assertions.assertEquals;
021import static org.junit.jupiter.api.Assertions.assertFalse;
022import static org.mockito.Mockito.mock;
023
024import java.io.IOException;
025import java.io.InterruptedIOException;
026import java.util.ArrayList;
027import java.util.Collection;
028import java.util.List;
029import java.util.concurrent.atomic.AtomicBoolean;
030import java.util.concurrent.atomic.AtomicReference;
031import org.apache.hadoop.conf.Configuration;
032import org.apache.hadoop.fs.FileSystem;
033import org.apache.hadoop.fs.Path;
034import org.apache.hadoop.hbase.HBaseTestingUtil;
035import org.apache.hadoop.hbase.Stoppable;
036import org.apache.hadoop.hbase.TableName;
037import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
038import org.apache.hadoop.hbase.client.Put;
039import org.apache.hadoop.hbase.client.RegionInfo;
040import org.apache.hadoop.hbase.client.RegionInfoBuilder;
041import org.apache.hadoop.hbase.client.TableDescriptor;
042import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
043import org.apache.hadoop.hbase.testclassification.RegionServerTests;
044import org.apache.hadoop.hbase.testclassification.SmallTests;
045import org.apache.hadoop.hbase.util.Bytes;
046import org.apache.hadoop.hbase.util.CommonFSUtils;
047import org.apache.hadoop.hbase.wal.WALFactory;
048import org.junit.jupiter.api.AfterEach;
049import org.junit.jupiter.api.BeforeEach;
050import org.junit.jupiter.api.Tag;
051import org.junit.jupiter.api.Test;
052import org.junit.jupiter.api.TestInfo;
053import org.mockito.Mockito;
054
055/**
056 * Tests a race condition between archiving of compacted files in CompactedHFilesDischarger chore
057 * and HRegion.close();
058 */
059@Tag(RegionServerTests.TAG)
060@Tag(SmallTests.TAG)
061public class TestCompactionArchiveConcurrentClose {
062
063  private HBaseTestingUtil testUtil;
064
065  private Path testDir;
066  private AtomicBoolean archived = new AtomicBoolean();
067
068  private String name;
069
070  @BeforeEach
071  public void setup(TestInfo testInfo) throws Exception {
072    this.name = testInfo.getTestMethod().get().getName();
073    testUtil = new HBaseTestingUtil();
074    testDir = testUtil.getDataTestDir("TestStoreFileRefresherChore");
075    CommonFSUtils.setRootDir(testUtil.getConfiguration(), testDir);
076  }
077
078  @AfterEach
079  public void tearDown() throws Exception {
080    testUtil.cleanupTestDir();
081  }
082
083  @Test
084  public void testStoreCloseAndDischargeRunningInParallel() throws Exception {
085    byte[] fam = Bytes.toBytes("f");
086    byte[] col = Bytes.toBytes("c");
087    byte[] val = Bytes.toBytes("val");
088
089    TableName tableName = TableName.valueOf(name);
090    TableDescriptor htd = TableDescriptorBuilder.newBuilder(tableName)
091      .setColumnFamily(ColumnFamilyDescriptorBuilder.of(fam)).build();
092    RegionInfo info = RegionInfoBuilder.newBuilder(tableName).build();
093    HRegion region = initHRegion(htd, info);
094    RegionServerServices rss = mock(RegionServerServices.class);
095    List<HRegion> regions = new ArrayList<>();
096    regions.add(region);
097    Mockito.doReturn(regions).when(rss).getRegions();
098
099    // Create the cleaner object
100    CompactedHFilesDischarger cleaner =
101      new CompactedHFilesDischarger(1000, (Stoppable) null, rss, false);
102    // Add some data to the region and do some flushes
103    int batchSize = 10;
104    int fileCount = 10;
105    for (int f = 0; f < fileCount; f++) {
106      int start = f * batchSize;
107      for (int i = start; i < start + batchSize; i++) {
108        Put p = new Put(Bytes.toBytes("row" + i));
109        p.addColumn(fam, col, val);
110        region.put(p);
111      }
112      // flush them
113      region.flush(true);
114    }
115
116    HStore store = region.getStore(fam);
117    assertEquals(fileCount, store.getStorefilesCount());
118
119    Collection<HStoreFile> storefiles = store.getStorefiles();
120    // None of the files should be in compacted state.
121    for (HStoreFile file : storefiles) {
122      assertFalse(file.isCompactedAway());
123    }
124    // Do compaction
125    region.compact(true);
126
127    // now run the cleaner with a concurrent close
128    Thread cleanerThread = new Thread() {
129      @Override
130      public void run() {
131        cleaner.chore();
132      }
133    };
134    cleanerThread.start();
135    // wait for cleaner to pause
136    synchronized (archived) {
137      if (!archived.get()) {
138        archived.wait();
139      }
140    }
141    final AtomicReference<Exception> closeException = new AtomicReference<>();
142    Thread closeThread = new Thread() {
143      @Override
144      public void run() {
145        // wait for the chore to complete and call close
146        try {
147          ((HRegion) region).close();
148        } catch (IOException e) {
149          closeException.set(e);
150        }
151      }
152    };
153    closeThread.start();
154    // no error should occur after the execution of the test
155    closeThread.join();
156    cleanerThread.join();
157
158    if (closeException.get() != null) {
159      throw closeException.get();
160    }
161  }
162
163  private HRegion initHRegion(TableDescriptor htd, RegionInfo info) throws IOException {
164    Configuration conf = testUtil.getConfiguration();
165    Path tableDir = CommonFSUtils.getTableDir(testDir, htd.getTableName());
166
167    HRegionFileSystem fs =
168      new WaitingHRegionFileSystem(conf, tableDir.getFileSystem(conf), tableDir, info);
169    ChunkCreator.initialize(MemStoreLAB.CHUNK_SIZE_DEFAULT, false, 0, 0, 0, null,
170      MemStoreLAB.INDEX_CHUNK_SIZE_PERCENTAGE_DEFAULT);
171    final Configuration walConf = new Configuration(conf);
172    CommonFSUtils.setRootDir(walConf, tableDir);
173    final WALFactory wals = new WALFactory(walConf, "log_" + info.getEncodedName());
174    HRegion region = new HRegion(fs, wals.getWAL(info), conf, htd, null);
175    Path regionDir = new Path(tableDir, info.getEncodedName());
176    if (!fs.getFileSystem().exists(regionDir)) {
177      fs.getFileSystem().mkdirs(regionDir);
178    }
179    region.initialize();
180    return region;
181  }
182
183  private class WaitingHRegionFileSystem extends HRegionFileSystem {
184
185    public WaitingHRegionFileSystem(final Configuration conf, final FileSystem fs,
186      final Path tableDir, final RegionInfo regionInfo) {
187      super(conf, fs, tableDir, regionInfo);
188    }
189
190    @Override
191    public void removeStoreFiles(String familyName, Collection<HStoreFile> storeFiles)
192      throws IOException {
193      super.removeStoreFiles(familyName, storeFiles);
194      archived.set(true);
195      synchronized (archived) {
196        archived.notifyAll();
197      }
198      try {
199        // unfortunately we can't use a stronger barrier here as the fix synchronizing
200        // the race condition will then block
201        Thread.sleep(100);
202      } catch (InterruptedException ie) {
203        throw new InterruptedIOException("Interrupted waiting for latch");
204      }
205    }
206  }
207}