View Javadoc

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  
19  package org.apache.hadoop.hbase.regionserver;
20  
21  import static java.util.Arrays.asList;
22  import static org.junit.Assert.assertEquals;
23  import static org.junit.Assert.assertNotNull;
24  import static org.junit.Assert.assertTrue;
25  
26  import java.io.File;
27  import java.io.FileNotFoundException;
28  import java.io.FileOutputStream;
29  import java.io.IOException;
30  import java.util.ArrayList;
31  import java.util.Arrays;
32  import java.util.List;
33  import java.util.Random;
34  
35  import org.apache.hadoop.conf.Configuration;
36  import org.apache.hadoop.fs.FSDataOutputStream;
37  import org.apache.hadoop.fs.Path;
38  import org.apache.hadoop.hbase.CellUtil;
39  import org.apache.hadoop.hbase.DoNotRetryIOException;
40  import org.apache.hadoop.hbase.HBaseConfiguration;
41  import org.apache.hadoop.hbase.HColumnDescriptor;
42  import org.apache.hadoop.hbase.HRegionInfo;
43  import org.apache.hadoop.hbase.HTableDescriptor;
44  import org.apache.hadoop.hbase.KeyValue;
45  import org.apache.hadoop.hbase.TableName;
46  import org.apache.hadoop.hbase.io.hfile.HFile;
47  import org.apache.hadoop.hbase.io.hfile.HFileContext;
48  import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
49  import org.apache.hadoop.hbase.protobuf.generated.WALProtos.BulkLoadDescriptor;
50  import org.apache.hadoop.hbase.protobuf.generated.WALProtos.StoreDescriptor;
51  import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
52  import org.apache.hadoop.hbase.testclassification.SmallTests;
53  import org.apache.hadoop.hbase.util.Bytes;
54  import org.apache.hadoop.hbase.util.Pair;
55  import org.apache.hadoop.hbase.wal.WAL;
56  import org.apache.hadoop.hbase.wal.WALKey;
57  import org.hamcrest.Description;
58  import org.hamcrest.Matcher;
59  import org.hamcrest.TypeSafeMatcher;
60  
61  import static org.mockito.Matchers.*;
62  import static org.mockito.Mockito.mock;
63  import static org.mockito.Mockito.verify;
64  import static org.mockito.Mockito.when;
65  import org.mockito.invocation.InvocationOnMock;
66  import org.mockito.stubbing.Answer;
67  
68  import org.junit.Before;
69  import org.junit.ClassRule;
70  import org.junit.Rule;
71  import org.junit.Test;
72  import org.junit.experimental.categories.Category;
73  import org.junit.rules.TemporaryFolder;
74  import org.junit.rules.TestName;
75  
76  /**
77   * This class attempts to unit test bulk HLog loading.
78   */
79  @Category(SmallTests.class)
80  public class TestBulkLoad {
81  
82    @ClassRule
83    public static TemporaryFolder testFolder = new TemporaryFolder();
84    private final WAL log = mock(WAL.class);
85    private final Configuration conf = HBaseConfiguration.create();
86    private final Random random = new Random();
87    private final byte[] randomBytes = new byte[100];
88    private final byte[] family1 = Bytes.toBytes("family1");
89    private final byte[] family2 = Bytes.toBytes("family2");
90    @Rule
91    public TestName name = new TestName();
92  
93    @Before
94    public void before() throws IOException {
95      random.nextBytes(randomBytes);
96      // Mockito.when(log.append(htd, info, key, edits, inMemstore));
97    }
98  
99    @Test
100   public void verifyBulkLoadEvent() throws IOException {
101     TableName tableName = TableName.valueOf("test", "test");
102     List<Pair<byte[], String>> familyPaths = withFamilyPathsFor(family1);
103     byte[] familyName = familyPaths.get(0).getFirst();
104     String storeFileName = familyPaths.get(0).getSecond();
105     storeFileName = (new Path(storeFileName)).getName();
106     List<String> storeFileNames = new ArrayList<String>();
107     storeFileNames.add(storeFileName);
108     when(log.append(any(HTableDescriptor.class), any(HRegionInfo.class), any(WALKey.class),
109             argThat(bulkLogWalEdit(WALEdit.BULK_LOAD, tableName.toBytes(),
110                     familyName, storeFileNames)),
111             any(boolean.class))).thenAnswer(new Answer() {
112       public Object answer(InvocationOnMock invocation) {
113         WALKey walKey = invocation.getArgumentAt(2, WALKey.class);
114         MultiVersionConcurrencyControl mvcc = walKey.getMvcc();
115         if (mvcc != null) {
116           MultiVersionConcurrencyControl.WriteEntry we = mvcc.begin();
117           walKey.setWriteEntry(we);
118         }
119         return 01L;
120       };
121     });
122     testRegionWithFamiliesAndSpecifiedTableName(tableName, family1)
123         .bulkLoadHFiles(familyPaths, false, null);
124     verify(log).sync(anyLong());
125   }
126 
127   @Test
128   public void bulkHLogShouldThrowNoErrorAndWriteMarkerWithBlankInput() throws IOException {
129     testRegionWithFamilies(family1).bulkLoadHFiles(new ArrayList<Pair<byte[], String>>(),
130       false, null);
131   }
132 
133   @Test
134   public void shouldBulkLoadSingleFamilyHLog() throws IOException {
135     when(log.append(any(HTableDescriptor.class), any(HRegionInfo.class),
136             any(WALKey.class), argThat(bulkLogWalEditType(WALEdit.BULK_LOAD)),
137             any(boolean.class))).thenAnswer(new Answer() {
138       public Object answer(InvocationOnMock invocation) {
139         WALKey walKey = invocation.getArgumentAt(2, WALKey.class);
140         MultiVersionConcurrencyControl mvcc = walKey.getMvcc();
141         if (mvcc != null) {
142           MultiVersionConcurrencyControl.WriteEntry we = mvcc.begin();
143           walKey.setWriteEntry(we);
144         }
145         return 01L;
146       };
147     });
148     testRegionWithFamilies(family1).bulkLoadHFiles(withFamilyPathsFor(family1), false, null);
149     verify(log).sync(anyLong());
150   }
151 
152   @Test
153   public void shouldBulkLoadManyFamilyHLog() throws IOException {
154     when(log.append(any(HTableDescriptor.class), any(HRegionInfo.class),
155             any(WALKey.class), argThat(bulkLogWalEditType(WALEdit.BULK_LOAD)),
156             any(boolean.class))).thenAnswer(new Answer() {
157               public Object answer(InvocationOnMock invocation) {
158                 WALKey walKey = invocation.getArgumentAt(2, WALKey.class);
159                 MultiVersionConcurrencyControl mvcc = walKey.getMvcc();
160                 if (mvcc != null) {
161                   MultiVersionConcurrencyControl.WriteEntry we = mvcc.begin();
162                   walKey.setWriteEntry(we);
163                 }
164                 return 01L;
165               };
166             });
167     testRegionWithFamilies(family1, family2).bulkLoadHFiles(withFamilyPathsFor(family1, family2),
168             false, null);
169     verify(log).sync(anyLong());
170   }
171 
172   @Test
173   public void shouldBulkLoadManyFamilyHLogEvenWhenTableNameNamespaceSpecified() throws IOException {
174     when(log.append(any(HTableDescriptor.class), any(HRegionInfo.class),
175             any(WALKey.class), argThat(bulkLogWalEditType(WALEdit.BULK_LOAD)),
176             any(boolean.class))).thenAnswer(new Answer() {
177       public Object answer(InvocationOnMock invocation) {
178         WALKey walKey = invocation.getArgumentAt(2, WALKey.class);
179         MultiVersionConcurrencyControl mvcc = walKey.getMvcc();
180         if (mvcc != null) {
181           MultiVersionConcurrencyControl.WriteEntry we = mvcc.begin();
182           walKey.setWriteEntry(we);
183         }
184         return 01L;
185       };
186     });
187     TableName tableName = TableName.valueOf("test", "test");
188     testRegionWithFamiliesAndSpecifiedTableName(tableName, family1, family2)
189         .bulkLoadHFiles(withFamilyPathsFor(family1, family2), false, null);
190     verify(log).sync(anyLong());
191   }
192 
193   @Test(expected = DoNotRetryIOException.class)
194   public void shouldCrashIfBulkLoadFamiliesNotInTable() throws IOException {
195     testRegionWithFamilies(family1).bulkLoadHFiles(withFamilyPathsFor(family1, family2), false,
196       null);
197   }
198 
199   @Test(expected = DoNotRetryIOException.class)
200   public void bulkHLogShouldThrowErrorWhenFamilySpecifiedAndHFileExistsButNotInTableDescriptor()
201       throws IOException {
202     testRegionWithFamilies().bulkLoadHFiles(withFamilyPathsFor(family1), false, null);
203   }
204 
205   @Test(expected = DoNotRetryIOException.class)
206   public void shouldThrowErrorIfBadFamilySpecifiedAsFamilyPath() throws IOException {
207     testRegionWithFamilies()
208         .bulkLoadHFiles(asList(withInvalidColumnFamilyButProperHFileLocation(family1)),
209             false, null);
210   }
211 
212   @Test(expected = FileNotFoundException.class)
213   public void shouldThrowErrorIfHFileDoesNotExist() throws IOException {
214     List<Pair<byte[], String>> list = asList(withMissingHFileForFamily(family1));
215     testRegionWithFamilies(family1).bulkLoadHFiles(list, false, null);
216   }
217 
218   private Pair<byte[], String> withMissingHFileForFamily(byte[] family) {
219     return new Pair<byte[], String>(family, "/tmp/does_not_exist");
220   }
221 
222   private Pair<byte[], String> withInvalidColumnFamilyButProperHFileLocation(byte[] family)
223       throws IOException {
224     createHFileForFamilies(family);
225     return new Pair<byte[], String>(new byte[]{0x00, 0x01, 0x02}, "/tmp/does_not_exist");
226   }
227 
228 
229   private HRegion testRegionWithFamiliesAndSpecifiedTableName(TableName tableName,
230                                                               byte[]... families)
231   throws IOException {
232     HRegionInfo hRegionInfo = new HRegionInfo(tableName);
233     HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName);
234     for (byte[] family : families) {
235       hTableDescriptor.addFamily(new HColumnDescriptor(family));
236     }
237 
238     // TODO We need a way to do this without creating files
239     return HRegion.createHRegion(hRegionInfo,
240         new Path(testFolder.newFolder().toURI()),
241         conf,
242         hTableDescriptor,
243         log);
244 
245   }
246 
247   private HRegion testRegionWithFamilies(byte[]... families) throws IOException {
248     TableName tableName = TableName.valueOf(name.getMethodName());
249     return testRegionWithFamiliesAndSpecifiedTableName(tableName, families);
250   }
251 
252   private List<Pair<byte[], String>> getBlankFamilyPaths(){
253     return new ArrayList<Pair<byte[], String>>();
254   }
255 
256   private List<Pair<byte[], String>> withFamilyPathsFor(byte[]... families) throws IOException {
257     List<Pair<byte[], String>> familyPaths = getBlankFamilyPaths();
258     for (byte[] family : families) {
259       familyPaths.add(new Pair<byte[], String>(family, createHFileForFamilies(family)));
260     }
261     return familyPaths;
262   }
263 
264   private String createHFileForFamilies(byte[] family) throws IOException {
265     HFile.WriterFactory hFileFactory = HFile.getWriterFactoryNoCache(conf);
266     // TODO We need a way to do this without creating files
267     File hFileLocation = testFolder.newFile();
268     FSDataOutputStream out = new FSDataOutputStream(new FileOutputStream(hFileLocation));
269     try {
270       hFileFactory.withOutputStream(out);
271       hFileFactory.withFileContext(new HFileContext());
272       HFile.Writer writer = hFileFactory.create();
273       try {
274         writer.append(new KeyValue(CellUtil.createCell(randomBytes,
275             family,
276             randomBytes,
277             0l,
278             KeyValue.Type.Put.getCode(),
279             randomBytes)));
280       } finally {
281         writer.close();
282       }
283     } finally {
284       out.close();
285     }
286     return hFileLocation.getAbsoluteFile().getAbsolutePath();
287   }
288 
289   private static Matcher<WALEdit> bulkLogWalEditType(byte[] typeBytes) {
290     return new WalMatcher(typeBytes);
291   }
292 
293   private static Matcher<WALEdit> bulkLogWalEdit(byte[] typeBytes, byte[] tableName,
294       byte[] familyName, List<String> storeFileNames) {
295     return new WalMatcher(typeBytes, tableName, familyName, storeFileNames);
296   }
297 
298   private static class WalMatcher extends TypeSafeMatcher<WALEdit> {
299     private final byte[] typeBytes;
300     private final byte[] tableName;
301     private final byte[] familyName;
302     private final List<String> storeFileNames;
303 
304     public WalMatcher(byte[] typeBytes) {
305       this(typeBytes, null, null, null);
306     }
307 
308     public WalMatcher(byte[] typeBytes, byte[] tableName, byte[] familyName,
309         List<String> storeFileNames) {
310       this.typeBytes = typeBytes;
311       this.tableName = tableName;
312       this.familyName = familyName;
313       this.storeFileNames = storeFileNames;
314     }
315 
316     @Override
317     protected boolean matchesSafely(WALEdit item) {
318       assertTrue(Arrays.equals(item.getCells().get(0).getQualifier(), typeBytes));
319       BulkLoadDescriptor desc;
320       try {
321         desc = WALEdit.getBulkLoadDescriptor(item.getCells().get(0));
322       } catch (IOException e) {
323         return false;
324       }
325       assertNotNull(desc);
326 
327       if (tableName != null) {
328         assertTrue(Bytes.equals(ProtobufUtil.toTableName(desc.getTableName()).getName(),
329           tableName));
330       }
331 
332       if(storeFileNames != null) {
333         int index=0;
334         StoreDescriptor store = desc.getStores(0);
335         assertTrue(Bytes.equals(store.getFamilyName().toByteArray(), familyName));
336         assertTrue(Bytes.equals(Bytes.toBytes(store.getStoreHomeDir()), familyName));
337         assertEquals(storeFileNames.size(), store.getStoreFileCount());
338       }
339 
340       return true;
341     }
342 
343     @Override
344     public void describeTo(Description description) {
345 
346     }
347   }
348 }