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.coprocessor;
019
020import static org.junit.Assert.assertEquals;
021import static org.junit.Assert.assertNull;
022
023import java.io.IOException;
024import java.util.List;
025import java.util.Optional;
026import java.util.concurrent.CountDownLatch;
027import org.apache.hadoop.conf.Configuration;
028import org.apache.hadoop.fs.FileSystem;
029import org.apache.hadoop.fs.Path;
030import org.apache.hadoop.hbase.Cell;
031import org.apache.hadoop.hbase.Coprocessor;
032import org.apache.hadoop.hbase.HBaseClassTestRule;
033import org.apache.hadoop.hbase.HBaseTestingUtility;
034import org.apache.hadoop.hbase.HColumnDescriptor;
035import org.apache.hadoop.hbase.HConstants;
036import org.apache.hadoop.hbase.HRegionInfo;
037import org.apache.hadoop.hbase.HTableDescriptor;
038import org.apache.hadoop.hbase.TableName;
039import org.apache.hadoop.hbase.client.Admin;
040import org.apache.hadoop.hbase.client.Get;
041import org.apache.hadoop.hbase.client.Put;
042import org.apache.hadoop.hbase.client.RegionInfo;
043import org.apache.hadoop.hbase.client.Result;
044import org.apache.hadoop.hbase.client.Scan;
045import org.apache.hadoop.hbase.client.Table;
046import org.apache.hadoop.hbase.client.TableDescriptor;
047import org.apache.hadoop.hbase.filter.FilterBase;
048import org.apache.hadoop.hbase.regionserver.ChunkCreator;
049import org.apache.hadoop.hbase.regionserver.FlushLifeCycleTracker;
050import org.apache.hadoop.hbase.regionserver.HRegion;
051import org.apache.hadoop.hbase.regionserver.HRegionServer;
052import org.apache.hadoop.hbase.regionserver.HStore;
053import org.apache.hadoop.hbase.regionserver.InternalScanner;
054import org.apache.hadoop.hbase.regionserver.MemStoreLAB;
055import org.apache.hadoop.hbase.regionserver.Region;
056import org.apache.hadoop.hbase.regionserver.RegionCoprocessorHost;
057import org.apache.hadoop.hbase.regionserver.RegionServerServices;
058import org.apache.hadoop.hbase.regionserver.ScanType;
059import org.apache.hadoop.hbase.regionserver.ScannerContext;
060import org.apache.hadoop.hbase.regionserver.Store;
061import org.apache.hadoop.hbase.regionserver.StoreScanner;
062import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext;
063import org.apache.hadoop.hbase.regionserver.compactions.CompactionLifeCycleTracker;
064import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequest;
065import org.apache.hadoop.hbase.regionserver.throttle.ThroughputController;
066import org.apache.hadoop.hbase.security.User;
067import org.apache.hadoop.hbase.testclassification.CoprocessorTests;
068import org.apache.hadoop.hbase.testclassification.MediumTests;
069import org.apache.hadoop.hbase.util.Bytes;
070import org.apache.hadoop.hbase.wal.WAL;
071import org.junit.ClassRule;
072import org.junit.Rule;
073import org.junit.Test;
074import org.junit.experimental.categories.Category;
075import org.junit.rules.TestName;
076
077@Category({ CoprocessorTests.class, MediumTests.class })
078public class TestRegionObserverScannerOpenHook {
079
080  @ClassRule
081  public static final HBaseClassTestRule CLASS_RULE =
082    HBaseClassTestRule.forClass(TestRegionObserverScannerOpenHook.class);
083
084  private static HBaseTestingUtility UTIL = new HBaseTestingUtility();
085  static final Path DIR = UTIL.getDataTestDir();
086
087  @Rule
088  public TestName name = new TestName();
089
090  public static class NoDataFilter extends FilterBase {
091
092    @Override
093    public ReturnCode filterCell(final Cell ignored) {
094      return ReturnCode.SKIP;
095    }
096
097    @Override
098    public boolean filterAllRemaining() throws IOException {
099      return true;
100    }
101
102    @Override
103    public boolean filterRow() throws IOException {
104      return true;
105    }
106  }
107
108  /**
109   * Do the default logic in {@link RegionObserver} interface.
110   */
111  public static class EmptyRegionObsever implements RegionCoprocessor, RegionObserver {
112    @Override
113    public Optional<RegionObserver> getRegionObserver() {
114      return Optional.of(this);
115    }
116  }
117
118  /**
119   * Don't return any data from a scan by creating a custom {@link StoreScanner}.
120   */
121  public static class NoDataFromScan implements RegionCoprocessor, RegionObserver {
122    @Override
123    public Optional<RegionObserver> getRegionObserver() {
124      return Optional.of(this);
125    }
126
127    @Override
128    public void preGetOp(ObserverContext<RegionCoprocessorEnvironment> c, Get get,
129      List<Cell> result) throws IOException {
130      c.bypass();
131    }
132
133    @Override
134    public void preScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Scan scan)
135      throws IOException {
136      scan.setFilter(new NoDataFilter());
137    }
138  }
139
140  private static final InternalScanner NO_DATA = new InternalScanner() {
141
142    @Override
143    public boolean next(List<Cell> result, ScannerContext scannerContext) throws IOException {
144      return false;
145    }
146
147    @Override
148    public void close() throws IOException {
149    }
150  };
151
152  /**
153   * Don't allow any data in a flush by creating a custom {@link StoreScanner}.
154   */
155  public static class NoDataFromFlush implements RegionCoprocessor, RegionObserver {
156    @Override
157    public Optional<RegionObserver> getRegionObserver() {
158      return Optional.of(this);
159    }
160
161    @Override
162    public InternalScanner preFlush(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
163      InternalScanner scanner, FlushLifeCycleTracker tracker) throws IOException {
164      return NO_DATA;
165    }
166  }
167
168  /**
169   * Don't allow any data to be written out in the compaction by creating a custom
170   * {@link StoreScanner}.
171   */
172  public static class NoDataFromCompaction implements RegionCoprocessor, RegionObserver {
173    @Override
174    public Optional<RegionObserver> getRegionObserver() {
175      return Optional.of(this);
176    }
177
178    @Override
179    public InternalScanner preCompact(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
180      InternalScanner scanner, ScanType scanType, CompactionLifeCycleTracker tracker,
181      CompactionRequest request) throws IOException {
182      return NO_DATA;
183    }
184  }
185
186  HRegion initHRegion(byte[] tableName, String callingMethod, Configuration conf,
187    byte[]... families) throws IOException {
188    HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(tableName));
189    for (byte[] family : families) {
190      htd.addFamily(new HColumnDescriptor(family));
191    }
192    ChunkCreator.initialize(MemStoreLAB.CHUNK_SIZE_DEFAULT, false, 0, 0, 0, null,
193      MemStoreLAB.INDEX_CHUNK_SIZE_PERCENTAGE_DEFAULT);
194    HRegionInfo info = new HRegionInfo(htd.getTableName(), null, null, false);
195    Path path = new Path(DIR + callingMethod);
196    WAL wal = HBaseTestingUtility.createWal(conf, path, info);
197    HRegion r = HRegion.createHRegion(info, path, conf, htd, wal);
198    // this following piece is a hack. currently a coprocessorHost
199    // is secretly loaded at OpenRegionHandler. we don't really
200    // start a region server here, so just manually create cphost
201    // and set it to region.
202    RegionCoprocessorHost host = new RegionCoprocessorHost(r, null, conf);
203    r.setCoprocessorHost(host);
204    return r;
205  }
206
207  @Test
208  public void testRegionObserverScanTimeStacking() throws Exception {
209    byte[] ROW = Bytes.toBytes("testRow");
210    byte[] TABLE = Bytes.toBytes(getClass().getName());
211    byte[] A = Bytes.toBytes("A");
212    byte[][] FAMILIES = new byte[][] { A };
213
214    // Use new HTU to not overlap with the DFS cluster started in #CompactionStacking
215    Configuration conf = new HBaseTestingUtility().getConfiguration();
216    HRegion region = initHRegion(TABLE, getClass().getName(), conf, FAMILIES);
217    RegionCoprocessorHost h = region.getCoprocessorHost();
218    h.load(NoDataFromScan.class, Coprocessor.PRIORITY_HIGHEST, conf);
219    h.load(EmptyRegionObsever.class, Coprocessor.PRIORITY_USER, conf);
220
221    Put put = new Put(ROW);
222    put.addColumn(A, A, A);
223    region.put(put);
224
225    Get get = new Get(ROW);
226    Result r = region.get(get);
227    assertNull(
228      "Got an unexpected number of rows - "
229        + "no data should be returned with the NoDataFromScan coprocessor. Found: " + r,
230      r.listCells());
231    HBaseTestingUtility.closeRegionAndWAL(region);
232  }
233
234  @Test
235  public void testRegionObserverFlushTimeStacking() throws Exception {
236    byte[] ROW = Bytes.toBytes("testRow");
237    byte[] TABLE = Bytes.toBytes(getClass().getName());
238    byte[] A = Bytes.toBytes("A");
239    byte[][] FAMILIES = new byte[][] { A };
240
241    // Use new HTU to not overlap with the DFS cluster started in #CompactionStacking
242    Configuration conf = new HBaseTestingUtility().getConfiguration();
243    HRegion region = initHRegion(TABLE, getClass().getName(), conf, FAMILIES);
244    RegionCoprocessorHost h = region.getCoprocessorHost();
245    h.load(NoDataFromFlush.class, Coprocessor.PRIORITY_HIGHEST, conf);
246    h.load(EmptyRegionObsever.class, Coprocessor.PRIORITY_USER, conf);
247
248    // put a row and flush it to disk
249    Put put = new Put(ROW);
250    put.addColumn(A, A, A);
251    region.put(put);
252    region.flush(true);
253    Get get = new Get(ROW);
254    Result r = region.get(get);
255    assertNull(
256      "Got an unexpected number of rows - "
257        + "no data should be returned with the NoDataFromScan coprocessor. Found: " + r,
258      r.listCells());
259    HBaseTestingUtility.closeRegionAndWAL(region);
260  }
261
262  /*
263   * Custom HRegion which uses CountDownLatch to signal the completion of compaction
264   */
265  public static class CompactionCompletionNotifyingRegion extends HRegion {
266    private static volatile CountDownLatch compactionStateChangeLatch = null;
267
268    @SuppressWarnings("deprecation")
269    public CompactionCompletionNotifyingRegion(Path tableDir, WAL log, FileSystem fs,
270      Configuration confParam, RegionInfo info, TableDescriptor htd,
271      RegionServerServices rsServices) {
272      super(tableDir, log, fs, confParam, info, htd, rsServices);
273    }
274
275    public CountDownLatch getCompactionStateChangeLatch() {
276      if (compactionStateChangeLatch == null) {
277        compactionStateChangeLatch = new CountDownLatch(1);
278      }
279      return compactionStateChangeLatch;
280    }
281
282    @Override
283    public boolean compact(CompactionContext compaction, HStore store,
284      ThroughputController throughputController) throws IOException {
285      boolean ret = super.compact(compaction, store, throughputController);
286      if (ret) {
287        compactionStateChangeLatch.countDown();
288      }
289      return ret;
290    }
291
292    @Override
293    public boolean compact(CompactionContext compaction, HStore store,
294      ThroughputController throughputController, User user) throws IOException {
295      boolean ret = super.compact(compaction, store, throughputController, user);
296      if (ret) compactionStateChangeLatch.countDown();
297      return ret;
298    }
299  }
300
301  /**
302   * Unfortunately, the easiest way to test this is to spin up a mini-cluster since we want to do
303   * the usual compaction mechanism on the region, rather than going through the backdoor to the
304   * region
305   */
306  @Test
307  public void testRegionObserverCompactionTimeStacking() throws Exception {
308    // setup a mini cluster so we can do a real compaction on a region
309    Configuration conf = UTIL.getConfiguration();
310    conf.setClass(HConstants.REGION_IMPL, CompactionCompletionNotifyingRegion.class, HRegion.class);
311    conf.setInt("hbase.hstore.compaction.min", 2);
312    UTIL.startMiniCluster();
313    byte[] ROW = Bytes.toBytes("testRow");
314    byte[] A = Bytes.toBytes("A");
315    HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(name.getMethodName()));
316    desc.addFamily(new HColumnDescriptor(A));
317    desc.addCoprocessor(EmptyRegionObsever.class.getName(), null, Coprocessor.PRIORITY_USER, null);
318    desc.addCoprocessor(NoDataFromCompaction.class.getName(), null, Coprocessor.PRIORITY_HIGHEST,
319      null);
320
321    Admin admin = UTIL.getAdmin();
322    admin.createTable(desc);
323
324    Table table = UTIL.getConnection().getTable(desc.getTableName());
325
326    // put a row and flush it to disk
327    Put put = new Put(ROW);
328    put.addColumn(A, A, A);
329    table.put(put);
330
331    HRegionServer rs = UTIL.getRSForFirstRegionInTable(desc.getTableName());
332    List<HRegion> regions = rs.getRegions(desc.getTableName());
333    assertEquals("More than 1 region serving test table with 1 row", 1, regions.size());
334    Region region = regions.get(0);
335    admin.flushRegion(region.getRegionInfo().getRegionName());
336    CountDownLatch latch =
337      ((CompactionCompletionNotifyingRegion) region).getCompactionStateChangeLatch();
338
339    // put another row and flush that too
340    put = new Put(Bytes.toBytes("anotherrow"));
341    put.addColumn(A, A, A);
342    table.put(put);
343    admin.flushRegion(region.getRegionInfo().getRegionName());
344
345    // run a compaction, which normally would should get rid of the data
346    // wait for the compaction checker to complete
347    latch.await();
348    // check both rows to ensure that they aren't there
349    Get get = new Get(ROW);
350    Result r = table.get(get);
351    assertNull(
352      "Got an unexpected number of rows - "
353        + "no data should be returned with the NoDataFromScan coprocessor. Found: " + r,
354      r.listCells());
355
356    get = new Get(Bytes.toBytes("anotherrow"));
357    r = table.get(get);
358    assertNull(
359      "Got an unexpected number of rows - "
360        + "no data should be returned with the NoDataFromScan coprocessor Found: " + r,
361      r.listCells());
362
363    table.close();
364    UTIL.shutdownMiniCluster();
365  }
366}