Source code

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.client;
019
020import java.io.IOException;
021import java.util.Arrays;
022import java.util.List;
023import org.apache.hadoop.conf.Configuration;
024import org.apache.hadoop.fs.FileSystem;
025import org.apache.hadoop.fs.Path;
026import org.apache.hadoop.hbase.Cell;
027import org.apache.hadoop.hbase.CellScanner;
028import org.apache.hadoop.hbase.HBaseClassTestRule;
029import org.apache.hadoop.hbase.HBaseTestingUtility;
030import org.apache.hadoop.hbase.HRegionInfo;
031import org.apache.hadoop.hbase.TableName;
032import org.apache.hadoop.hbase.master.snapshot.SnapshotManager;
033import org.apache.hadoop.hbase.snapshot.RestoreSnapshotHelper;
034import org.apache.hadoop.hbase.snapshot.SnapshotTestingUtils;
035import org.apache.hadoop.hbase.testclassification.ClientTests;
036import org.apache.hadoop.hbase.testclassification.LargeTests;
037import org.apache.hadoop.hbase.util.Bytes;
038import org.apache.hadoop.hbase.util.FSUtils;
039import org.junit.After;
040import org.junit.Assert;
041import org.junit.ClassRule;
042import org.junit.Test;
043import org.junit.experimental.categories.Category;
044import org.slf4j.Logger;
045import org.slf4j.LoggerFactory;
046
047@Category({LargeTests.class, ClientTests.class})
048public class TestTableSnapshotScanner {
049
050  @ClassRule
051  public static final HBaseClassTestRule CLASS_RULE =
052      HBaseClassTestRule.forClass(TestTableSnapshotScanner.class);
053
054  private static final Logger LOG = LoggerFactory.getLogger(TestTableSnapshotScanner.class);
055  private final HBaseTestingUtility UTIL = new HBaseTestingUtility();
056  private static final int NUM_REGION_SERVERS = 2;
057  private static final byte[][] FAMILIES = {Bytes.toBytes("f1"), Bytes.toBytes("f2")};
058  public static byte[] bbb = Bytes.toBytes("bbb");
059  public static byte[] yyy = Bytes.toBytes("yyy");
060
061  private FileSystem fs;
062  private Path rootDir;
063
064  public static void blockUntilSplitFinished(HBaseTestingUtility util, TableName tableName,
065      int expectedRegionSize) throws Exception {
066    for (int i = 0; i < 100; i++) {
067      List<HRegionInfo> hRegionInfoList = util.getAdmin().getTableRegions(tableName);
068      if (hRegionInfoList.size() >= expectedRegionSize) {
069        break;
070      }
071      Thread.sleep(1000);
072    }
073  }
074
075  public void setupCluster() throws Exception {
076    setupConf(UTIL.getConfiguration());
077    UTIL.startMiniCluster(NUM_REGION_SERVERS, true);
078    rootDir = UTIL.getHBaseCluster().getMaster().getMasterFileSystem().getRootDir();
079    fs = rootDir.getFileSystem(UTIL.getConfiguration());
080  }
081
082  public void tearDownCluster() throws Exception {
083    UTIL.shutdownMiniCluster();
084  }
085
086  private static void setupConf(Configuration conf) {
087    // Enable snapshot
088    conf.setBoolean(SnapshotManager.HBASE_SNAPSHOT_ENABLED, true);
089  }
090
091  @After
092  public void tearDown() throws Exception {
093  }
094
095  public static void createTableAndSnapshot(HBaseTestingUtility util, TableName tableName,
096      String snapshotName, int numRegions)
097      throws Exception {
098    try {
099      util.deleteTable(tableName);
100    } catch(Exception ex) {
101      // ignore
102    }
103
104    if (numRegions > 1) {
105      util.createTable(tableName, FAMILIES, 1, bbb, yyy, numRegions);
106    } else {
107      util.createTable(tableName, FAMILIES);
108    }
109    Admin admin = util.getAdmin();
110
111    // put some stuff in the table
112    Table table = util.getConnection().getTable(tableName);
113    util.loadTable(table, FAMILIES);
114
115    Path rootDir = FSUtils.getRootDir(util.getConfiguration());
116    FileSystem fs = rootDir.getFileSystem(util.getConfiguration());
117
118    SnapshotTestingUtils.createSnapshotAndValidate(admin, tableName,
119        Arrays.asList(FAMILIES), null, snapshotName, rootDir, fs, true);
120
121    // load different values
122    byte[] value = Bytes.toBytes("after_snapshot_value");
123    util.loadTable(table, FAMILIES, value);
124
125    // cause flush to create new files in the region
126    admin.flush(tableName);
127    table.close();
128  }
129
130  @Test
131  public void testNoDuplicateResultsWhenSplitting() throws Exception {
132    setupCluster();
133    TableName tableName = TableName.valueOf("testNoDuplicateResultsWhenSplitting");
134    String snapshotName = "testSnapshotBug";
135    try {
136      if (UTIL.getAdmin().tableExists(tableName)) {
137        UTIL.deleteTable(tableName);
138      }
139
140      UTIL.createTable(tableName, FAMILIES);
141      Admin admin = UTIL.getAdmin();
142
143      // put some stuff in the table
144      Table table = UTIL.getConnection().getTable(tableName);
145      UTIL.loadTable(table, FAMILIES);
146
147      // split to 2 regions
148      admin.split(tableName, Bytes.toBytes("eee"));
149      blockUntilSplitFinished(UTIL, tableName, 2);
150
151      Path rootDir = FSUtils.getRootDir(UTIL.getConfiguration());
152      FileSystem fs = rootDir.getFileSystem(UTIL.getConfiguration());
153
154      SnapshotTestingUtils.createSnapshotAndValidate(admin, tableName,
155        Arrays.asList(FAMILIES), null, snapshotName, rootDir, fs, true);
156
157      // load different values
158      byte[] value = Bytes.toBytes("after_snapshot_value");
159      UTIL.loadTable(table, FAMILIES, value);
160
161      // cause flush to create new files in the region
162      admin.flush(tableName);
163      table.close();
164
165      Path restoreDir = UTIL.getDataTestDirOnTestFS(snapshotName);
166      Scan scan = new Scan().withStartRow(bbb).withStopRow(yyy); // limit the scan
167
168      TableSnapshotScanner scanner =
169          new TableSnapshotScanner(UTIL.getConfiguration(), restoreDir, snapshotName, scan);
170
171      verifyScanner(scanner, bbb, yyy);
172      scanner.close();
173    } finally {
174      UTIL.getAdmin().deleteSnapshot(snapshotName);
175      UTIL.deleteTable(tableName);
176      tearDownCluster();
177    }
178  }
179
180  @Test
181  public void testWithSingleRegion() throws Exception {
182    testScanner(UTIL, "testWithSingleRegion", 1, false);
183  }
184
185  @Test
186  public void testWithMultiRegion() throws Exception {
187    testScanner(UTIL, "testWithMultiRegion", 10, false);
188  }
189
190  @Test
191  public void testWithOfflineHBaseMultiRegion() throws Exception {
192    testScanner(UTIL, "testWithMultiRegion", 20, true);
193  }
194
195  @Test
196  public void testScannerWithRestoreScanner() throws Exception {
197    setupCluster();
198    TableName tableName = TableName.valueOf("testScanner");
199    String snapshotName = "testScannerWithRestoreScanner";
200    try {
201      createTableAndSnapshot(UTIL, tableName, snapshotName, 50);
202      Path restoreDir = UTIL.getDataTestDirOnTestFS(snapshotName);
203      Scan scan = new Scan(bbb, yyy); // limit the scan
204
205      Configuration conf = UTIL.getConfiguration();
206      Path rootDir = FSUtils.getRootDir(conf);
207
208      TableSnapshotScanner scanner0 =
209          new TableSnapshotScanner(conf, restoreDir, snapshotName, scan);
210      verifyScanner(scanner0, bbb, yyy);
211      scanner0.close();
212
213      // restore snapshot.
214      RestoreSnapshotHelper.copySnapshotForScanner(conf, fs, rootDir, restoreDir, snapshotName);
215
216      // scan the snapshot without restoring snapshot
217      TableSnapshotScanner scanner =
218          new TableSnapshotScanner(conf, rootDir, restoreDir, snapshotName, scan, true);
219      verifyScanner(scanner, bbb, yyy);
220      scanner.close();
221
222      // check whether the snapshot has been deleted by the close of scanner.
223      scanner = new TableSnapshotScanner(conf, rootDir, restoreDir, snapshotName, scan, true);
224      verifyScanner(scanner, bbb, yyy);
225      scanner.close();
226
227      // restore snapshot again.
228      RestoreSnapshotHelper.copySnapshotForScanner(conf, fs, rootDir, restoreDir, snapshotName);
229
230      // check whether the snapshot has been deleted by the close of scanner.
231      scanner = new TableSnapshotScanner(conf, rootDir, restoreDir, snapshotName, scan, true);
232      verifyScanner(scanner, bbb, yyy);
233      scanner.close();
234    } finally {
235      UTIL.getAdmin().deleteSnapshot(snapshotName);
236      UTIL.deleteTable(tableName);
237      tearDownCluster();
238    }
239  }
240
241  private void testScanner(HBaseTestingUtility util, String snapshotName, int numRegions,
242      boolean shutdownCluster) throws Exception {
243    setupCluster();
244    TableName tableName = TableName.valueOf("testScanner");
245    try {
246      createTableAndSnapshot(util, tableName, snapshotName, numRegions);
247
248      if (shutdownCluster) {
249        util.shutdownMiniHBaseCluster();
250      }
251
252      Path restoreDir = util.getDataTestDirOnTestFS(snapshotName);
253      Scan scan = new Scan(bbb, yyy); // limit the scan
254
255      TableSnapshotScanner scanner = new TableSnapshotScanner(UTIL.getConfiguration(), restoreDir,
256        snapshotName, scan);
257
258      verifyScanner(scanner, bbb, yyy);
259      scanner.close();
260    } finally {
261      if (!shutdownCluster) {
262        util.getAdmin().deleteSnapshot(snapshotName);
263        util.deleteTable(tableName);
264        tearDownCluster();
265      }
266    }
267  }
268
269  private void verifyScanner(ResultScanner scanner, byte[] startRow, byte[] stopRow)
270      throws IOException, InterruptedException {
271
272    HBaseTestingUtility.SeenRowTracker rowTracker =
273        new HBaseTestingUtility.SeenRowTracker(startRow, stopRow);
274
275    while (true) {
276      Result result = scanner.next();
277      if (result == null) {
278        break;
279      }
280      verifyRow(result);
281      rowTracker.addRow(result.getRow());
282    }
283
284    // validate all rows are seen
285    rowTracker.validate();
286  }
287
288  private static void verifyRow(Result result) throws IOException {
289    byte[] row = result.getRow();
290    CellScanner scanner = result.cellScanner();
291    while (scanner.advance()) {
292      Cell cell = scanner.current();
293
294      //assert that all Cells in the Result have the same key
295     Assert.assertEquals(0, Bytes.compareTo(row, 0, row.length,
296         cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()));
297    }
298
299    for (int j = 0; j < FAMILIES.length; j++) {
300      byte[] actual = result.getValue(FAMILIES[j], FAMILIES[j]);
301      Assert.assertArrayEquals("Row in snapshot does not match, expected:" + Bytes.toString(row)
302          + " ,actual:" + Bytes.toString(actual), row, actual);
303    }
304  }
305
306}