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.wal;
019
020import static org.junit.Assert.assertEquals;
021import static org.junit.Assert.assertFalse;
022import static org.junit.Assert.assertNull;
023import static org.junit.Assert.assertTrue;
024
025import java.io.IOException;
026import java.util.HashSet;
027import java.util.NavigableMap;
028import java.util.Set;
029import java.util.TreeMap;
030import java.util.concurrent.ThreadLocalRandom;
031import org.apache.hadoop.conf.Configuration;
032import org.apache.hadoop.fs.FileStatus;
033import org.apache.hadoop.fs.FileSystem;
034import org.apache.hadoop.fs.Path;
035import org.apache.hadoop.hbase.HBaseClassTestRule;
036import org.apache.hadoop.hbase.HBaseTestingUtil;
037import org.apache.hadoop.hbase.HConstants;
038import org.apache.hadoop.hbase.KeyValue;
039import org.apache.hadoop.hbase.ServerName;
040import org.apache.hadoop.hbase.TableName;
041import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
042import org.apache.hadoop.hbase.client.RegionInfo;
043import org.apache.hadoop.hbase.client.RegionInfoBuilder;
044import org.apache.hadoop.hbase.client.TableDescriptor;
045import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
046import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl;
047import org.apache.hadoop.hbase.testclassification.MediumTests;
048import org.apache.hadoop.hbase.testclassification.RegionServerTests;
049import org.apache.hadoop.hbase.util.Bytes;
050import org.apache.hadoop.hbase.util.CommonFSUtils;
051import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
052import org.junit.AfterClass;
053import org.junit.Before;
054import org.junit.BeforeClass;
055import org.junit.ClassRule;
056import org.junit.Rule;
057import org.junit.Test;
058import org.junit.experimental.categories.Category;
059import org.junit.rules.TestName;
060import org.slf4j.Logger;
061import org.slf4j.LoggerFactory;
062
063@Category({ RegionServerTests.class, MediumTests.class })
064public class TestFSHLogProvider {
065
066  @ClassRule
067  public static final HBaseClassTestRule CLASS_RULE =
068    HBaseClassTestRule.forClass(TestFSHLogProvider.class);
069
070  private static final Logger LOG = LoggerFactory.getLogger(TestFSHLogProvider.class);
071
072  private static Configuration conf;
073  private static FileSystem fs;
074  private final static HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
075  private MultiVersionConcurrencyControl mvcc;
076
077  @Rule
078  public final TestName currentTest = new TestName();
079
080  @Before
081  public void setUp() throws Exception {
082    mvcc = new MultiVersionConcurrencyControl();
083    FileStatus[] entries = fs.listStatus(new Path("/"));
084    for (FileStatus dir : entries) {
085      fs.delete(dir.getPath(), true);
086    }
087  }
088
089  @BeforeClass
090  public static void setUpBeforeClass() throws Exception {
091    // Make block sizes small.
092    TEST_UTIL.getConfiguration().setInt("dfs.blocksize", 1024 * 1024);
093    // quicker heartbeat interval for faster DN death notification
094    TEST_UTIL.getConfiguration().setInt("dfs.namenode.heartbeat.recheck-interval", 5000);
095    TEST_UTIL.getConfiguration().setInt("dfs.heartbeat.interval", 1);
096    TEST_UTIL.getConfiguration().setInt("dfs.client.socket-timeout", 5000);
097
098    // faster failover with cluster.shutdown();fs.close() idiom
099    TEST_UTIL.getConfiguration().setInt("hbase.ipc.client.connect.max.retries", 1);
100    TEST_UTIL.getConfiguration().setInt("dfs.client.block.recovery.retries", 1);
101    TEST_UTIL.getConfiguration().setInt("hbase.ipc.client.connection.maxidletime", 500);
102    TEST_UTIL.startMiniDFSCluster(3);
103
104    // Set up a working space for our tests.
105    TEST_UTIL.createRootDir();
106    conf = TEST_UTIL.getConfiguration();
107    fs = TEST_UTIL.getDFSCluster().getFileSystem();
108  }
109
110  @AfterClass
111  public static void tearDownAfterClass() throws Exception {
112    TEST_UTIL.shutdownMiniCluster();
113  }
114
115  static String getName() {
116    return "TestDefaultWALProvider";
117  }
118
119  @Test
120  public void testGetServerNameFromWALDirectoryName() throws IOException {
121    ServerName sn = ServerName.valueOf("hn", 450, 1398);
122    String hl = CommonFSUtils.getRootDir(conf) + "/"
123      + AbstractFSWALProvider.getWALDirectoryName(sn.toString());
124
125    // Must not throw exception
126    assertNull(AbstractFSWALProvider.getServerNameFromWALDirectoryName(conf, null));
127    assertNull(AbstractFSWALProvider.getServerNameFromWALDirectoryName(conf,
128      CommonFSUtils.getRootDir(conf).toUri().toString()));
129    assertNull(AbstractFSWALProvider.getServerNameFromWALDirectoryName(conf, ""));
130    assertNull(AbstractFSWALProvider.getServerNameFromWALDirectoryName(conf, "                  "));
131    assertNull(AbstractFSWALProvider.getServerNameFromWALDirectoryName(conf, hl));
132    assertNull(AbstractFSWALProvider.getServerNameFromWALDirectoryName(conf, hl + "qdf"));
133    assertNull(AbstractFSWALProvider.getServerNameFromWALDirectoryName(conf, "sfqf" + hl + "qdf"));
134
135    final String wals = "/WALs/";
136    ServerName parsed = AbstractFSWALProvider.getServerNameFromWALDirectoryName(conf,
137      CommonFSUtils.getRootDir(conf).toUri().toString() + wals + sn
138        + "/localhost%2C32984%2C1343316388997.1343316390417");
139    assertEquals("standard", sn, parsed);
140
141    parsed = AbstractFSWALProvider.getServerNameFromWALDirectoryName(conf, hl + "/qdf");
142    assertEquals("subdir", sn, parsed);
143
144    parsed = AbstractFSWALProvider.getServerNameFromWALDirectoryName(conf,
145      CommonFSUtils.getRootDir(conf).toUri().toString() + wals + sn
146        + "-splitting/localhost%3A57020.1340474893931");
147    assertEquals("split", sn, parsed);
148  }
149
150  private void addEdits(WAL log, RegionInfo hri, TableDescriptor htd, int times,
151    NavigableMap<byte[], Integer> scopes) throws IOException {
152    final byte[] row = Bytes.toBytes("row");
153    for (int i = 0; i < times; i++) {
154      long timestamp = EnvironmentEdgeManager.currentTime();
155      WALEdit cols = new WALEdit();
156      cols.add(new KeyValue(row, row, row, timestamp, row));
157      log.appendData(hri,
158        getWalKey(hri.getEncodedNameAsBytes(), htd.getTableName(), timestamp, scopes), cols);
159    }
160    log.sync();
161  }
162
163  /**
164   * used by TestDefaultWALProviderWithHLogKey n
165   */
166  WALKeyImpl getWalKey(final byte[] info, final TableName tableName, final long timestamp,
167    NavigableMap<byte[], Integer> scopes) {
168    return new WALKeyImpl(info, tableName, timestamp, mvcc, scopes);
169  }
170
171  /**
172   * helper method to simulate region flush for a WAL. nn
173   */
174  protected void flushRegion(WAL wal, byte[] regionEncodedName, Set<byte[]> flushedFamilyNames) {
175    wal.startCacheFlush(regionEncodedName, flushedFamilyNames);
176    wal.completeCacheFlush(regionEncodedName, HConstants.NO_SEQNUM);
177  }
178
179  @Test
180  public void testLogCleaning() throws Exception {
181    LOG.info(currentTest.getMethodName());
182    TableDescriptor htd =
183      TableDescriptorBuilder.newBuilder(TableName.valueOf(currentTest.getMethodName()))
184        .setColumnFamily(ColumnFamilyDescriptorBuilder.of("row")).build();
185    TableDescriptor htd2 =
186      TableDescriptorBuilder.newBuilder(TableName.valueOf(currentTest.getMethodName() + "2"))
187        .setColumnFamily(ColumnFamilyDescriptorBuilder.of("row")).build();
188    NavigableMap<byte[], Integer> scopes1 = new TreeMap<>(Bytes.BYTES_COMPARATOR);
189    for (byte[] fam : htd.getColumnFamilyNames()) {
190      scopes1.put(fam, 0);
191    }
192    NavigableMap<byte[], Integer> scopes2 = new TreeMap<>(Bytes.BYTES_COMPARATOR);
193    for (byte[] fam : htd2.getColumnFamilyNames()) {
194      scopes2.put(fam, 0);
195    }
196    Configuration localConf = new Configuration(conf);
197    localConf.set(WALFactory.WAL_PROVIDER, FSHLogProvider.class.getName());
198    WALFactory wals = new WALFactory(localConf, currentTest.getMethodName());
199    try {
200      RegionInfo hri = RegionInfoBuilder.newBuilder(htd.getTableName()).build();
201      RegionInfo hri2 = RegionInfoBuilder.newBuilder(htd2.getTableName()).build();
202      // we want to mix edits from regions, so pick our own identifier.
203      WAL log = wals.getWAL(null);
204
205      // Add a single edit and make sure that rolling won't remove the file
206      // Before HBASE-3198 it used to delete it
207      addEdits(log, hri, htd, 1, scopes1);
208      log.rollWriter();
209      assertEquals(1, AbstractFSWALProvider.getNumRolledLogFiles(log));
210
211      // See if there's anything wrong with more than 1 edit
212      addEdits(log, hri, htd, 2, scopes1);
213      log.rollWriter();
214      assertEquals(2, FSHLogProvider.getNumRolledLogFiles(log));
215
216      // Now mix edits from 2 regions, still no flushing
217      addEdits(log, hri, htd, 1, scopes1);
218      addEdits(log, hri2, htd2, 1, scopes2);
219      addEdits(log, hri, htd, 1, scopes1);
220      addEdits(log, hri2, htd2, 1, scopes2);
221      log.rollWriter();
222      assertEquals(3, AbstractFSWALProvider.getNumRolledLogFiles(log));
223
224      // Flush the first region, we expect to see the first two files getting
225      // archived. We need to append something or writer won't be rolled.
226      addEdits(log, hri2, htd2, 1, scopes2);
227      log.startCacheFlush(hri.getEncodedNameAsBytes(), htd.getColumnFamilyNames());
228      log.completeCacheFlush(hri.getEncodedNameAsBytes(), HConstants.NO_SEQNUM);
229      log.rollWriter();
230      int count = AbstractFSWALProvider.getNumRolledLogFiles(log);
231      assertEquals(2, count);
232
233      // Flush the second region, which removes all the remaining output files
234      // since the oldest was completely flushed and the two others only contain
235      // flush information
236      addEdits(log, hri2, htd2, 1, scopes2);
237      log.startCacheFlush(hri2.getEncodedNameAsBytes(), htd2.getColumnFamilyNames());
238      log.completeCacheFlush(hri2.getEncodedNameAsBytes(), HConstants.NO_SEQNUM);
239      log.rollWriter();
240      assertEquals(0, AbstractFSWALProvider.getNumRolledLogFiles(log));
241    } finally {
242      if (wals != null) {
243        wals.close();
244      }
245    }
246  }
247
248  /**
249   * Tests wal archiving by adding data, doing flushing/rolling and checking we archive old logs and
250   * also don't archive "live logs" (that is, a log with un-flushed entries).
251   * <p>
252   * This is what it does: It creates two regions, and does a series of inserts along with log
253   * rolling. Whenever a WAL is rolled, HLogBase checks previous wals for archiving. A wal is
254   * eligible for archiving if for all the regions which have entries in that wal file, have flushed
255   * - past their maximum sequence id in that wal file.
256   * <p>
257   * n
258   */
259  @Test
260  public void testWALArchiving() throws IOException {
261    LOG.debug(currentTest.getMethodName());
262    TableDescriptor table1 =
263      TableDescriptorBuilder.newBuilder(TableName.valueOf(currentTest.getMethodName() + "1"))
264        .setColumnFamily(ColumnFamilyDescriptorBuilder.of("row")).build();
265    TableDescriptor table2 =
266      TableDescriptorBuilder.newBuilder(TableName.valueOf(currentTest.getMethodName() + "2"))
267        .setColumnFamily(ColumnFamilyDescriptorBuilder.of("row")).build();
268    NavigableMap<byte[], Integer> scopes1 = new TreeMap<>(Bytes.BYTES_COMPARATOR);
269    for (byte[] fam : table1.getColumnFamilyNames()) {
270      scopes1.put(fam, 0);
271    }
272    NavigableMap<byte[], Integer> scopes2 = new TreeMap<>(Bytes.BYTES_COMPARATOR);
273    for (byte[] fam : table2.getColumnFamilyNames()) {
274      scopes2.put(fam, 0);
275    }
276    Configuration localConf = new Configuration(conf);
277    localConf.set(WALFactory.WAL_PROVIDER, FSHLogProvider.class.getName());
278    WALFactory wals = new WALFactory(localConf, currentTest.getMethodName());
279    try {
280      WAL wal = wals.getWAL(null);
281      assertEquals(0, AbstractFSWALProvider.getNumRolledLogFiles(wal));
282      RegionInfo hri1 = RegionInfoBuilder.newBuilder(table1.getTableName()).build();
283      RegionInfo hri2 = RegionInfoBuilder.newBuilder(table2.getTableName()).build();
284      // variables to mock region sequenceIds.
285      // start with the testing logic: insert a waledit, and roll writer
286      addEdits(wal, hri1, table1, 1, scopes1);
287      wal.rollWriter();
288      // assert that the wal is rolled
289      assertEquals(1, AbstractFSWALProvider.getNumRolledLogFiles(wal));
290      // add edits in the second wal file, and roll writer.
291      addEdits(wal, hri1, table1, 1, scopes1);
292      wal.rollWriter();
293      // assert that the wal is rolled
294      assertEquals(2, AbstractFSWALProvider.getNumRolledLogFiles(wal));
295      // add a waledit to table1, and flush the region.
296      addEdits(wal, hri1, table1, 3, scopes1);
297      flushRegion(wal, hri1.getEncodedNameAsBytes(), table1.getColumnFamilyNames());
298      // roll log; all old logs should be archived.
299      wal.rollWriter();
300      assertEquals(0, AbstractFSWALProvider.getNumRolledLogFiles(wal));
301      // add an edit to table2, and roll writer
302      addEdits(wal, hri2, table2, 1, scopes2);
303      wal.rollWriter();
304      assertEquals(1, AbstractFSWALProvider.getNumRolledLogFiles(wal));
305      // add edits for table1, and roll writer
306      addEdits(wal, hri1, table1, 2, scopes1);
307      wal.rollWriter();
308      assertEquals(2, AbstractFSWALProvider.getNumRolledLogFiles(wal));
309      // add edits for table2, and flush hri1.
310      addEdits(wal, hri2, table2, 2, scopes2);
311      flushRegion(wal, hri1.getEncodedNameAsBytes(), table2.getColumnFamilyNames());
312      // the log : region-sequenceId map is
313      // log1: region2 (unflushed)
314      // log2: region1 (flushed)
315      // log3: region2 (unflushed)
316      // roll the writer; log2 should be archived.
317      wal.rollWriter();
318      assertEquals(2, AbstractFSWALProvider.getNumRolledLogFiles(wal));
319      // flush region2, and all logs should be archived.
320      addEdits(wal, hri2, table2, 2, scopes2);
321      flushRegion(wal, hri2.getEncodedNameAsBytes(), table2.getColumnFamilyNames());
322      wal.rollWriter();
323      assertEquals(0, AbstractFSWALProvider.getNumRolledLogFiles(wal));
324    } finally {
325      if (wals != null) {
326        wals.close();
327      }
328    }
329  }
330
331  /**
332   * Write to a log file with three concurrent threads and verifying all data is written. n
333   */
334  @Test
335  public void testConcurrentWrites() throws Exception {
336    // Run the WPE tool with three threads writing 3000 edits each concurrently.
337    // When done, verify that all edits were written.
338    int errCode =
339      WALPerformanceEvaluation.innerMain(new Configuration(TEST_UTIL.getConfiguration()),
340        new String[] { "-threads", "3", "-verify", "-noclosefs", "-iterations", "3000" });
341    assertEquals(0, errCode);
342  }
343
344  /**
345   * Ensure that we can use Set.add to deduplicate WALs
346   */
347  @Test
348  public void setMembershipDedups() throws IOException {
349    Configuration localConf = new Configuration(conf);
350    localConf.set(WALFactory.WAL_PROVIDER, FSHLogProvider.class.getName());
351    WALFactory wals = new WALFactory(localConf, currentTest.getMethodName());
352    try {
353      final Set<WAL> seen = new HashSet<>(1);
354      assertTrue("first attempt to add WAL from default provider should work.",
355        seen.add(wals.getWAL(null)));
356      for (int i = 0; i < 1000; i++) {
357        assertFalse(
358          "default wal provider is only supposed to return a single wal, which should "
359            + "compare as .equals itself.",
360          seen.add(wals.getWAL(RegionInfoBuilder
361            .newBuilder(TableName.valueOf("Table-" + ThreadLocalRandom.current().nextInt()))
362            .build())));
363      }
364    } finally {
365      wals.close();
366    }
367  }
368}