hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestFlushLifeCycleTracker.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.assertFalse;
  21 import static org.junit.Assert.assertNotNull;
  22 import static org.junit.Assert.assertNull;
  23 import static org.junit.Assert.assertSame;
  24 import static org.junit.Assert.assertTrue;
  25
  26 import java.io.IOException;
  27 import java.io.InterruptedIOException;
  28 import java.util.Optional;
  29 import java.util.concurrent.CountDownLatch;
  30 import org.apache.hadoop.hbase.Cell.Type;
  31 import org.apache.hadoop.hbase.CellBuilderFactory;
  32 import org.apache.hadoop.hbase.CellBuilderType;
  33 import org.apache.hadoop.hbase.HBaseClassTestRule;
  34 import org.apache.hadoop.hbase.HBaseTestingUtil;
  35 import org.apache.hadoop.hbase.HConstants;
  36 import org.apache.hadoop.hbase.TableName;
  37 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
  38 import org.apache.hadoop.hbase.client.Put;
  39 import org.apache.hadoop.hbase.client.Table;
  40 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
  41 import org.apache.hadoop.hbase.coprocessor.ObserverContext;
  42 import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
  43 import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
  44 import org.apache.hadoop.hbase.coprocessor.RegionObserver;
  45 import org.apache.hadoop.hbase.testclassification.CoprocessorTests;
  46 import org.apache.hadoop.hbase.testclassification.MediumTests;
  47 import org.apache.hadoop.hbase.util.Bytes;
  48 import org.junit.After;
  49 import org.junit.AfterClass;
  50 import org.junit.Before;
  51 import org.junit.BeforeClass;
  52 import org.junit.ClassRule;
  53 import org.junit.Test;
  54 import org.junit.experimental.categories.Category;
  55
  56 /**
  57  * Confirm that the function of FlushLifeCycleTracker is OK as we do not use it in our own code.
  58  */
  59 @Category({ CoprocessorTests.class, MediumTests.class })
  60 public class TestFlushLifeCycleTracker {
  61
  62   @ClassRule
  63   public static final HBaseClassTestRule CLASS_RULE =
  64       HBaseClassTestRule.forClass(TestFlushLifeCycleTracker.class);
  65
  66   private static final HBaseTestingUtil UTIL = new HBaseTestingUtil();
  67
  68   private static final TableName NAME =
  69       TableName.valueOf(TestFlushLifeCycleTracker.class.getSimpleName());
  70
  71   private static final byte[] CF = Bytes.toBytes("CF");
  72
  73   private static final byte[] QUALIFIER = Bytes.toBytes("CQ");
  74
  75   private HRegion region;
  76
  77   private static FlushLifeCycleTracker TRACKER;
  78
  79   private static volatile CountDownLatch ARRIVE;
  80
  81   private static volatile CountDownLatch BLOCK;
  82
  83   public static final class FlushObserver implements RegionObserver, RegionCoprocessor {
  84
  85     @Override
  86     public Optional<RegionObserver> getRegionObserver() {
  87       return Optional.of(this);
  88     }
  89
  90     @Override
  91     public void preFlush(ObserverContext<RegionCoprocessorEnvironment> c,
  92         FlushLifeCycleTracker tracker) throws IOException {
  93       if (TRACKER != null) {
  94         assertSame(tracker, TRACKER);
  95       }
  96     }
  97
  98     @Override
  99     public InternalScanner preFlush(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
 100         InternalScanner scanner, FlushLifeCycleTracker tracker) throws IOException {
 101       if (TRACKER != null) {
 102         assertSame(tracker, TRACKER);
 103       }
 104       return scanner;
 105     }
 106
 107     @Override
 108     public void postFlush(ObserverContext<RegionCoprocessorEnvironment> c,
 109         FlushLifeCycleTracker tracker) throws IOException {
 110       if (TRACKER != null) {
 111         assertSame(tracker, TRACKER);
 112       }
 113     }
 114
 115     @Override
 116     public void postFlush(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
 117         StoreFile resultFile, FlushLifeCycleTracker tracker) throws IOException {
 118       if (TRACKER != null) {
 119         assertSame(tracker, TRACKER);
 120       }
 121       // inject here so we can make a flush request to fail because of we already have a flush
 122       // ongoing.
 123       CountDownLatch arrive = ARRIVE;
 124       if (arrive != null) {
 125         arrive.countDown();
 126         try {
 127           BLOCK.await();
 128         } catch (InterruptedException e) {
 129           throw new InterruptedIOException();
 130         }
 131       }
 132     }
 133   }
 134
 135   private static final class Tracker implements FlushLifeCycleTracker {
 136
 137     private String reason;
 138
 139     private boolean beforeExecutionCalled;
 140
 141     private boolean afterExecutionCalled;
 142
 143     private boolean completed = false;
 144
 145     @Override
 146     public synchronized void notExecuted(String reason) {
 147       this.reason = reason;
 148       completed = true;
 149       notifyAll();
 150     }
 151
 152     @Override
 153     public void beforeExecution() {
 154       this.beforeExecutionCalled = true;
 155     }
 156
 157     @Override
 158     public synchronized void afterExecution() {
 159       this.afterExecutionCalled = true;
 160       completed = true;
 161       notifyAll();
 162     }
 163
 164     public synchronized void await() throws InterruptedException {
 165       while (!completed) {
 166         wait();
 167       }
 168     }
 169   }
 170
 171   @BeforeClass
 172   public static void setUpBeforeClass() throws Exception {
 173     UTIL.startMiniCluster(3);
 174   }
 175
 176   @AfterClass
 177   public static void tearDownAfterClass() throws Exception {
 178     UTIL.shutdownMiniCluster();
 179   }
 180
 181   @Before
 182   public void setUp() throws IOException {
 183     UTIL.getAdmin()
 184         .createTable(TableDescriptorBuilder.newBuilder(NAME)
 185             .setColumnFamily(ColumnFamilyDescriptorBuilder.of(CF))
 186             .setCoprocessor(FlushObserver.class.getName()).build());
 187     region = UTIL.getHBaseCluster().getRegions(NAME).get(0);
 188   }
 189
 190   @After
 191   public void tearDown() throws IOException {
 192     region = null;
 193     TRACKER = null;
 194     UTIL.deleteTable(NAME);
 195   }
 196
 197   @Test
 198   public void test() throws IOException, InterruptedException {
 199     try (Table table = UTIL.getConnection().getTable(NAME)) {
 200       for (int i = 0; i < 100; i++) {
 201         byte[] row = Bytes.toBytes(i);
 202         table.put(new Put(row, true)
 203                     .add(CellBuilderFactory.create(CellBuilderType.SHALLOW_COPY)
 204                         .setRow(row)
 205                         .setFamily(CF)
 206                         .setQualifier(QUALIFIER)
 207                         .setTimestamp(HConstants.LATEST_TIMESTAMP)
 208                         .setType(Type.Put)
 209                         .setValue(Bytes.toBytes(i))
 210                         .build()));
 211       }
 212     }
 213     Tracker tracker = new Tracker();
 214     TRACKER = tracker;
 215     region.requestFlush(tracker);
 216     tracker.await();
 217     assertNull(tracker.reason);
 218     assertTrue(tracker.beforeExecutionCalled);
 219     assertTrue(tracker.afterExecutionCalled);
 220
 221     // request flush on a region with empty memstore should still success
 222     tracker = new Tracker();
 223     TRACKER = tracker;
 224     region.requestFlush(tracker);
 225     tracker.await();
 226     assertNull(tracker.reason);
 227     assertTrue(tracker.beforeExecutionCalled);
 228     assertTrue(tracker.afterExecutionCalled);
 229   }
 230
 231   @Test
 232   public void testNotExecuted() throws IOException, InterruptedException {
 233     try (Table table = UTIL.getConnection().getTable(NAME)) {
 234       for (int i = 0; i < 100; i++) {
 235         byte[] row = Bytes.toBytes(i);
 236         table.put(new Put(row, true)
 237                     .add(CellBuilderFactory.create(CellBuilderType.SHALLOW_COPY)
 238                         .setRow(row)
 239                         .setFamily(CF)
 240                         .setQualifier(QUALIFIER)
 241                         .setTimestamp(HConstants.LATEST_TIMESTAMP)
 242                         .setType(Type.Put)
 243                         .setValue(Bytes.toBytes(i))
 244                         .build()));
 245       }
 246     }
 247     // here we may have overlap when calling the CP hooks so we do not assert on TRACKER
 248     Tracker tracker1 = new Tracker();
 249     ARRIVE = new CountDownLatch(1);
 250     BLOCK = new CountDownLatch(1);
 251     region.requestFlush(tracker1);
 252     ARRIVE.await();
 253
 254     Tracker tracker2 = new Tracker();
 255     region.requestFlush(tracker2);
 256     tracker2.await();
 257     assertNotNull(tracker2.reason);
 258     assertFalse(tracker2.beforeExecutionCalled);
 259     assertFalse(tracker2.afterExecutionCalled);
 260
 261     BLOCK.countDown();
 262     tracker1.await();
 263     assertNull(tracker1.reason);
 264     assertTrue(tracker1.beforeExecutionCalled);
 265     assertTrue(tracker1.afterExecutionCalled);
 266   }
 267 }