View Javadoc

1   /*
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.replication.regionserver;
20  
21  import static org.junit.Assert.assertEquals;
22  import static org.junit.Assert.assertTrue;
23  
24  import java.security.SecureRandom;
25  import java.util.ArrayList;
26  import java.util.Collections;
27  import java.util.HashMap;
28  import java.util.HashSet;
29  import java.util.Iterator;
30  import java.util.List;
31  import java.util.Map;
32  import java.util.Random;
33  import java.util.Set;
34  import java.util.concurrent.atomic.AtomicBoolean;
35  
36  import org.apache.hadoop.hbase.TableNotFoundException;
37  import org.apache.hadoop.hbase.client.Admin;
38  import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
39  import org.apache.hadoop.hbase.client.Table;
40  import org.apache.hadoop.hbase.util.ByteStringer;
41  import org.apache.commons.logging.Log;
42  import org.apache.commons.logging.LogFactory;
43  import org.apache.hadoop.conf.Configuration;
44  import org.apache.hadoop.fs.FileSystem;
45  import org.apache.hadoop.fs.FileUtil;
46  import org.apache.hadoop.fs.Path;
47  import org.apache.hadoop.hbase.Cell;
48  import org.apache.hadoop.hbase.CellUtil;
49  import org.apache.hadoop.hbase.HBaseTestingUtility;
50  import org.apache.hadoop.hbase.HConstants;
51  import org.apache.hadoop.hbase.HRegionInfo;
52  import org.apache.hadoop.hbase.KeyValue;
53  import org.apache.hadoop.hbase.testclassification.MediumTests;
54  import org.apache.hadoop.hbase.Stoppable;
55  import org.apache.hadoop.hbase.TableName;
56  import org.apache.hadoop.hbase.client.Connection;
57  import org.apache.hadoop.hbase.client.ConnectionFactory;
58  import org.apache.hadoop.hbase.client.Get;
59  import org.apache.hadoop.hbase.client.RegionLocator;
60  import org.apache.hadoop.hbase.client.Result;
61  import org.apache.hadoop.hbase.client.ResultScanner;
62  import org.apache.hadoop.hbase.client.Scan;
63  import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
64  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.WALEntry;
65  import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.UUID;
66  import org.apache.hadoop.hbase.protobuf.generated.WALProtos;
67  import org.apache.hadoop.hbase.protobuf.generated.WALProtos.WALKey;
68  import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
69  import org.apache.hadoop.hbase.util.Bytes;
70  import org.apache.hadoop.hbase.util.FSUtils;
71  import org.apache.hadoop.hbase.util.HFileTestUtil;
72  import org.junit.AfterClass;
73  import org.junit.Assert;
74  import org.junit.Before;
75  import org.junit.BeforeClass;
76  import org.junit.Test;
77  import org.junit.experimental.categories.Category;
78  
79  @Category(MediumTests.class)
80  public class TestReplicationSink {
81    private static final Log LOG = LogFactory.getLog(TestReplicationSink.class);
82    private static final int BATCH_SIZE = 10;
83  
84    protected final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
85  
86    protected static ReplicationSink SINK;
87  
88    protected static final TableName TABLE_NAME1 = TableName.valueOf("table1");
89    protected static final TableName TABLE_NAME2 = TableName.valueOf("table2");
90  
91    protected static final byte[] FAM_NAME1 = Bytes.toBytes("info1");
92    protected static final byte[] FAM_NAME2 = Bytes.toBytes("info2");
93  
94    protected static Table table1;
95    protected static Stoppable STOPPABLE = new Stoppable() {
96      final AtomicBoolean stop = new AtomicBoolean(false);
97  
98      @Override
99      public boolean isStopped() {
100       return this.stop.get();
101     }
102 
103     @Override
104     public void stop(String why) {
105       LOG.info("STOPPING BECAUSE: " + why);
106       this.stop.set(true);
107     }
108 
109   };
110 
111   protected static Table table2;
112   protected static String baseNamespaceDir;
113   protected static String hfileArchiveDir;
114   protected static String replicationClusterId;
115 
116    /**
117    * @throws java.lang.Exception
118    */
119   @BeforeClass
120   public static void setUpBeforeClass() throws Exception {
121     TEST_UTIL.getConfiguration().setBoolean("dfs.support.append", true);
122     TEST_UTIL.getConfiguration().setBoolean(HConstants.REPLICATION_ENABLE_KEY,
123         HConstants.REPLICATION_ENABLE_DEFAULT);
124     TEST_UTIL.getConfiguration().set("hbase.replication.source.fs.conf.provider",
125       TestSourceFSConfigurationProvider.class.getCanonicalName());
126 
127     TEST_UTIL.startMiniCluster(3);
128     SINK =
129       new ReplicationSink(new Configuration(TEST_UTIL.getConfiguration()), STOPPABLE);
130     table1 = TEST_UTIL.createTable(TABLE_NAME1, FAM_NAME1);
131     table2 = TEST_UTIL.createTable(TABLE_NAME2, FAM_NAME2);
132     Path rootDir = FSUtils.getRootDir(TEST_UTIL.getConfiguration());
133     baseNamespaceDir = new Path(rootDir, new Path(HConstants.BASE_NAMESPACE_DIR)).toString();
134     hfileArchiveDir = new Path(rootDir, new Path(HConstants.HFILE_ARCHIVE_DIRECTORY)).toString();
135     replicationClusterId = "12345";
136   }
137 
138   /**
139    * @throws java.lang.Exception
140    */
141   @AfterClass
142   public static void tearDownAfterClass() throws Exception {
143     STOPPABLE.stop("Shutting down");
144     TEST_UTIL.shutdownMiniCluster();
145   }
146 
147   /**
148    * @throws java.lang.Exception
149    */
150   @Before
151   public void setUp() throws Exception {
152     table1 = TEST_UTIL.deleteTableData(TABLE_NAME1);
153     table2 = TEST_UTIL.deleteTableData(TABLE_NAME2);
154   }
155 
156   /**
157    * Insert a whole batch of entries
158    * @throws Exception
159    */
160   @Test
161   public void testBatchSink() throws Exception {
162     List<WALEntry> entries = new ArrayList<WALEntry>(BATCH_SIZE);
163     List<Cell> cells = new ArrayList<Cell>();
164     for(int i = 0; i < BATCH_SIZE; i++) {
165       entries.add(createEntry(TABLE_NAME1, i, KeyValue.Type.Put, cells));
166     }
167     SINK.replicateEntries(entries, CellUtil.createCellScanner(cells.iterator()),
168       replicationClusterId, baseNamespaceDir, hfileArchiveDir);
169     Scan scan = new Scan();
170     ResultScanner scanRes = table1.getScanner(scan);
171     assertEquals(BATCH_SIZE, scanRes.next(BATCH_SIZE).length);
172   }
173 
174   /**
175    * Insert a mix of puts and deletes
176    * @throws Exception
177    */
178   @Test
179   public void testMixedPutDelete() throws Exception {
180     List<WALEntry> entries = new ArrayList<WALEntry>(BATCH_SIZE/2);
181     List<Cell> cells = new ArrayList<Cell>();
182     for(int i = 0; i < BATCH_SIZE/2; i++) {
183       entries.add(createEntry(TABLE_NAME1, i, KeyValue.Type.Put, cells));
184     }
185     SINK.replicateEntries(entries, CellUtil.createCellScanner(cells), replicationClusterId,
186       baseNamespaceDir, hfileArchiveDir);
187 
188     entries = new ArrayList<WALEntry>(BATCH_SIZE);
189     cells = new ArrayList<Cell>();
190     for(int i = 0; i < BATCH_SIZE; i++) {
191       entries.add(createEntry(TABLE_NAME1, i,
192           i % 2 != 0 ? KeyValue.Type.Put: KeyValue.Type.DeleteColumn, cells));
193     }
194 
195     SINK.replicateEntries(entries, CellUtil.createCellScanner(cells.iterator()),
196       replicationClusterId, baseNamespaceDir, hfileArchiveDir);
197     Scan scan = new Scan();
198     ResultScanner scanRes = table1.getScanner(scan);
199     assertEquals(BATCH_SIZE/2, scanRes.next(BATCH_SIZE).length);
200   }
201 
202   /**
203    * Insert to 2 different tables
204    * @throws Exception
205    */
206   @Test
207   public void testMixedPutTables() throws Exception {
208     List<WALEntry> entries = new ArrayList<WALEntry>(BATCH_SIZE/2);
209     List<Cell> cells = new ArrayList<Cell>();
210     for(int i = 0; i < BATCH_SIZE; i++) {
211       entries.add(createEntry( i % 2 == 0 ? TABLE_NAME2 : TABLE_NAME1,
212               i, KeyValue.Type.Put, cells));
213     }
214 
215     SINK.replicateEntries(entries, CellUtil.createCellScanner(cells.iterator()),
216       replicationClusterId, baseNamespaceDir, hfileArchiveDir);
217     Scan scan = new Scan();
218     ResultScanner scanRes = table2.getScanner(scan);
219     for(Result res : scanRes) {
220       assertTrue(Bytes.toInt(res.getRow()) % 2 == 0);
221     }
222   }
223 
224   /**
225    * Insert then do different types of deletes
226    * @throws Exception
227    */
228   @Test
229   public void testMixedDeletes() throws Exception {
230     List<WALEntry> entries = new ArrayList<WALEntry>(3);
231     List<Cell> cells = new ArrayList<Cell>();
232     for(int i = 0; i < 3; i++) {
233       entries.add(createEntry(TABLE_NAME1, i, KeyValue.Type.Put, cells));
234     }
235     SINK.replicateEntries(entries, CellUtil.createCellScanner(cells.iterator()),
236       replicationClusterId, baseNamespaceDir, hfileArchiveDir);
237     entries = new ArrayList<WALEntry>(3);
238     cells = new ArrayList<Cell>();
239     entries.add(createEntry(TABLE_NAME1, 0, KeyValue.Type.DeleteColumn, cells));
240     entries.add(createEntry(TABLE_NAME1, 1, KeyValue.Type.DeleteFamily, cells));
241     entries.add(createEntry(TABLE_NAME1, 2, KeyValue.Type.DeleteColumn, cells));
242 
243     SINK.replicateEntries(entries, CellUtil.createCellScanner(cells.iterator()),
244       replicationClusterId, baseNamespaceDir, hfileArchiveDir);
245 
246     Scan scan = new Scan();
247     ResultScanner scanRes = table1.getScanner(scan);
248     assertEquals(0, scanRes.next(3).length);
249   }
250 
251   /**
252    * Puts are buffered, but this tests when a delete (not-buffered) is applied
253    * before the actual Put that creates it.
254    * @throws Exception
255    */
256   @Test
257   public void testApplyDeleteBeforePut() throws Exception {
258     List<WALEntry> entries = new ArrayList<WALEntry>(5);
259     List<Cell> cells = new ArrayList<Cell>();
260     for(int i = 0; i < 2; i++) {
261       entries.add(createEntry(TABLE_NAME1, i, KeyValue.Type.Put, cells));
262     }
263     entries.add(createEntry(TABLE_NAME1, 1, KeyValue.Type.DeleteFamily, cells));
264     for(int i = 3; i < 5; i++) {
265       entries.add(createEntry(TABLE_NAME1, i, KeyValue.Type.Put, cells));
266     }
267     SINK.replicateEntries(entries, CellUtil.createCellScanner(cells.iterator()),
268       replicationClusterId, baseNamespaceDir, hfileArchiveDir);
269     Get get = new Get(Bytes.toBytes(1));
270     Result res = table1.get(get);
271     assertEquals(0, res.size());
272   }
273 
274   @Test
275   public void testRethrowRetriesExhaustedWithDetailsException() throws Exception {
276     TableName notExistTable = TableName.valueOf("notExistTable");
277     List<WALEntry> entries = new ArrayList<>();
278     List<Cell> cells = new ArrayList<>();
279     for (int i = 0; i < 10; i++) {
280       entries.add(createEntry(notExistTable, i, KeyValue.Type.Put, cells));
281     }
282     try {
283       SINK.replicateEntries(entries, CellUtil.createCellScanner(cells.iterator()),
284         replicationClusterId, baseNamespaceDir, hfileArchiveDir);
285       Assert.fail("Should re-throw TableNotFoundException.");
286     } catch (TableNotFoundException e) {
287     }
288     entries.clear();
289     cells.clear();
290     for (int i = 0; i < 10; i++) {
291       entries.add(createEntry(TABLE_NAME1, i, KeyValue.Type.Put, cells));
292     }
293     try (Connection conn = ConnectionFactory.createConnection(TEST_UTIL.getConfiguration())) {
294       try (Admin admin = conn.getAdmin()) {
295         admin.disableTable(TABLE_NAME1);
296         try {
297           SINK.replicateEntries(entries, CellUtil.createCellScanner(cells.iterator()),
298             replicationClusterId, baseNamespaceDir, hfileArchiveDir);
299           Assert.fail("Should re-throw RetriesExhaustedWithDetailsException.");
300         } catch (RetriesExhaustedWithDetailsException e) {
301         } finally {
302           admin.enableTable(TABLE_NAME1);
303         }
304       }
305     }
306   }
307 
308   /**
309    * Test replicateEntries with a bulk load entry for 25 HFiles
310    */
311   @Test
312   public void testReplicateEntriesForHFiles() throws Exception {
313     Path dir = TEST_UTIL.getDataTestDirOnTestFS("testReplicateEntries");
314     Path familyDir = new Path(dir, Bytes.toString(FAM_NAME1));
315     int numRows = 10;
316 
317     List<Path> p = new ArrayList<>(1);
318 
319     // 1. Generate 25 hfile ranges
320     Random rng = new SecureRandom();
321     Set<Integer> numbers = new HashSet<>();
322     while (numbers.size() < 50) {
323       numbers.add(rng.nextInt(1000));
324     }
325     List<Integer> numberList = new ArrayList<>(numbers);
326     Collections.sort(numberList);
327     Map<String, Long> storeFilesSize = new HashMap<String, Long>(1);
328 
329     // 2. Create 25 hfiles
330     Configuration conf = TEST_UTIL.getConfiguration();
331     FileSystem fs = dir.getFileSystem(conf);
332     Iterator<Integer> numbersItr = numberList.iterator();
333     for (int i = 0; i < 25; i++) {
334       Path hfilePath = new Path(familyDir, "hfile_" + i);
335       HFileTestUtil.createHFile(conf, fs, hfilePath, FAM_NAME1, FAM_NAME1,
336         Bytes.toBytes(numbersItr.next()), Bytes.toBytes(numbersItr.next()), numRows);
337       p.add(hfilePath);
338       storeFilesSize.put(hfilePath.getName(), fs.getFileStatus(hfilePath).getLen());
339     }
340 
341     // 3. Create a BulkLoadDescriptor and a WALEdit
342     Map<byte[], List<Path>> storeFiles = new HashMap<>(1);
343     storeFiles.put(FAM_NAME1, p);
344     WALEdit edit = null;
345     WALProtos.BulkLoadDescriptor loadDescriptor = null;
346 
347     try (Connection c = ConnectionFactory.createConnection(conf);
348         RegionLocator l = c.getRegionLocator(TABLE_NAME1)) {
349       HRegionInfo regionInfo = l.getAllRegionLocations().get(0).getRegionInfo();
350       loadDescriptor =
351           ProtobufUtil.toBulkLoadDescriptor(TABLE_NAME1,
352             ByteStringer.wrap(regionInfo.getEncodedNameAsBytes()), storeFiles, storeFilesSize, 1);
353       edit = WALEdit.createBulkLoadEvent(regionInfo, loadDescriptor);
354     }
355     List<WALEntry> entries = new ArrayList<WALEntry>(1);
356 
357     // 4. Create a WALEntryBuilder
358     WALEntry.Builder builder = createWALEntryBuilder(TABLE_NAME1);
359 
360     // 5. Copy the hfile to the path as it is in reality
361     for (int i = 0; i < 25; i++) {
362       String pathToHfileFromNS =
363           new StringBuilder(100).append(TABLE_NAME1.getNamespaceAsString()).append(Path.SEPARATOR)
364               .append(Bytes.toString(TABLE_NAME1.getName())).append(Path.SEPARATOR)
365               .append(Bytes.toString(loadDescriptor.getEncodedRegionName().toByteArray()))
366               .append(Path.SEPARATOR).append(Bytes.toString(FAM_NAME1)).append(Path.SEPARATOR)
367               .append("hfile_" + i).toString();
368       String dst = baseNamespaceDir + Path.SEPARATOR + pathToHfileFromNS;
369 
370       FileUtil.copy(fs, p.get(0), fs, new Path(dst), false, conf);
371     }
372 
373     entries.add(builder.build());
374     try (ResultScanner scanner = table1.getScanner(new Scan())) {
375       // 6. Assert no existing data in table
376       assertEquals(0, scanner.next(numRows).length);
377     }
378     // 7. Replicate the bulk loaded entry
379     SINK.replicateEntries(entries, CellUtil.createCellScanner(edit.getCells().iterator()),
380       replicationClusterId, baseNamespaceDir, hfileArchiveDir);
381     try (ResultScanner scanner = table1.getScanner(new Scan())) {
382       // 8. Assert data is replicated
383       assertEquals(numRows, scanner.next(numRows).length);
384     }
385   }
386 
387   private WALEntry createEntry(TableName table, int row,  KeyValue.Type type, List<Cell> cells) {
388     byte[] fam = table.equals(TABLE_NAME1) ? FAM_NAME1 : FAM_NAME2;
389     byte[] rowBytes = Bytes.toBytes(row);
390     // Just make sure we don't get the same ts for two consecutive rows with
391     // same key
392     try {
393       Thread.sleep(1);
394     } catch (InterruptedException e) {
395       LOG.info("Was interrupted while sleep, meh", e);
396     }
397     final long now = System.currentTimeMillis();
398     KeyValue kv = null;
399     if(type.getCode() == KeyValue.Type.Put.getCode()) {
400       kv = new KeyValue(rowBytes, fam, fam, now,
401           KeyValue.Type.Put, Bytes.toBytes(row));
402     } else if (type.getCode() == KeyValue.Type.DeleteColumn.getCode()) {
403         kv = new KeyValue(rowBytes, fam, fam,
404             now, KeyValue.Type.DeleteColumn);
405     } else if (type.getCode() == KeyValue.Type.DeleteFamily.getCode()) {
406         kv = new KeyValue(rowBytes, fam, null,
407             now, KeyValue.Type.DeleteFamily);
408     }
409     WALEntry.Builder builder = createWALEntryBuilder(table);
410     cells.add(kv);
411 
412     return builder.build();
413   }
414 
415   private WALEntry.Builder createWALEntryBuilder(TableName table) {
416     WALEntry.Builder builder = WALEntry.newBuilder();
417     builder.setAssociatedCellCount(1);
418     WALKey.Builder keyBuilder = WALKey.newBuilder();
419     UUID.Builder uuidBuilder = UUID.newBuilder();
420     uuidBuilder.setLeastSigBits(HConstants.DEFAULT_CLUSTER_ID.getLeastSignificantBits());
421     uuidBuilder.setMostSigBits(HConstants.DEFAULT_CLUSTER_ID.getMostSignificantBits());
422     keyBuilder.setClusterId(uuidBuilder.build());
423     keyBuilder.setTableName(ByteStringer.wrap(table.getName()));
424     keyBuilder.setWriteTime(System.currentTimeMillis());
425     keyBuilder.setEncodedRegionName(ByteStringer.wrap(HConstants.EMPTY_BYTE_ARRAY));
426     keyBuilder.setLogSequenceNumber(-1);
427     builder.setKey(keyBuilder.build());
428     return builder;
429   }
430 }